Skip to content

armnet-core API Reference

armnet_core

Shared wire types for the armnet platform.

Single source of truth for everything that crosses a component boundary (orchestrator HTTP API request/response bodies, NATS message envelopes, runtime SDK return values).

API_KEY_ENV module-attribute 

API_KEY_ENV = 'ARMNET_API_KEY'

API_KEY_HEADER module-attribute 

API_KEY_HEADER = 'X-Armnet-Api-Key'

DEFAULT_EMBODIMENT module-attribute 

DEFAULT_EMBODIMENT: Embodiment = 'lerobot/so-101'

DEFAULT_TASK module-attribute 

DEFAULT_TASK: Task = 'assemble_block_tower'

Embodiment module-attribute 

Embodiment = str

Task module-attribute 

Task = str

TELEOP_EVENT_NEXT_EPISODE module-attribute 

TELEOP_EVENT_NEXT_EPISODE = 'next_episode'

TELEOP_EVENT_RERECORD_EPISODE module-attribute 

TELEOP_EVENT_RERECORD_EPISODE = 'rerecord_episode'

TELEOP_EVENT_STOP_RECORDING module-attribute 

TELEOP_EVENT_STOP_RECORDING = 'stop_recording'

TELEOP_EVENTS module-attribute 

TELEOP_EVENTS = (TELEOP_EVENT_NEXT_EPISODE, TELEOP_EVENT_RERECORD_EPISODE, TELEOP_EVENT_STOP_RECORDING)

TerminalStatus module-attribute 

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

CellRuntimeStatus

Bases: str, Enum

Current availability state for a robot cell.

Source code in core/src/armnet_core/models.py 
56
57
58
59
60
61
class CellRuntimeStatus(str, Enum):
    """Current availability state for a robot cell."""

    AVAILABLE = "available"
    OCCUPIED = "occupied"
    OFFLINE = "offline"

AVAILABLE class-attribute instance-attribute 

AVAILABLE = 'available'

OCCUPIED class-attribute instance-attribute 

OCCUPIED = 'occupied'

OFFLINE class-attribute instance-attribute 

OFFLINE = 'offline'

CellStatus

Bases: BaseModel

Current status for one cell as reported by recent heartbeats.

Source code in core/src/armnet_core/models.py 
323
324
325
326
327
328
329
330
331
class CellStatus(BaseModel):
    """Current status for one cell as reported by recent heartbeats."""

    cell_id: str
    embodiment: Embodiment
    task: Task
    status: CellRuntimeStatus
    current_job_id: Optional[str] = None
    last_heartbeat_at: Optional[datetime] = None

cell_id instance-attribute 

cell_id: str

embodiment instance-attribute 

embodiment: Embodiment

task instance-attribute 

task: Task

status instance-attribute 

status: CellRuntimeStatus

current_job_id class-attribute instance-attribute 

current_job_id: Optional[str] = None

last_heartbeat_at class-attribute instance-attribute 

last_heartbeat_at: Optional[datetime] = None

CellStatusSummary

Bases: BaseModel

Aggregate availability for an embodiment, optionally scoped to a task.

task is None when the summary aggregates across every task of the embodiment (used to answer "is any cell of this embodiment available?" for task-less jobs).

Source code in core/src/armnet_core/models.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
class CellStatusSummary(BaseModel):
    """Aggregate availability for an embodiment, optionally scoped to a task.

    ``task`` is ``None`` when the summary aggregates across every task of the
    embodiment (used to answer "is any cell of this embodiment available?" for
    task-less jobs).
    """

    embodiment: Embodiment
    task: Optional[Task] = None
    status: CellRuntimeStatus
    cells: list[CellStatus] = Field(default_factory=list)
    heartbeat_timeout_seconds: int = 20

embodiment instance-attribute 

embodiment: Embodiment

task class-attribute instance-attribute 

task: Optional[Task] = None

status instance-attribute 

status: CellRuntimeStatus

cells class-attribute instance-attribute 

cells: list[CellStatus] = Field(default_factory=list)

heartbeat_timeout_seconds class-attribute instance-attribute 

heartbeat_timeout_seconds: int = 20

EmbodimentInfo

Bases: BaseModel

One known embodiment, as stored in the orchestrator database.

Source code in core/src/armnet_core/models.py 
590
591
592
593
594
class EmbodimentInfo(BaseModel):
    """One known embodiment, as stored in the orchestrator database."""

    slug: Embodiment = Field(..., description="Wire-format embodiment slug, e.g. 'lerobot/so-101'.")
    description: Optional[str] = None

slug class-attribute instance-attribute 

slug: Embodiment = Field(..., description="Wire-format embodiment slug, e.g. 'lerobot/so-101'.")

description class-attribute instance-attribute 

description: Optional[str] = None

EmbodimentList

Bases: BaseModel

The set of embodiments the platform currently accepts (GET /embodiments).

Source code in core/src/armnet_core/models.py 
597
598
599
600
class EmbodimentList(BaseModel):
    """The set of embodiments the platform currently accepts (GET /embodiments)."""

    embodiments: list[EmbodimentInfo] = Field(default_factory=list)

embodiments class-attribute instance-attribute 

embodiments: list[EmbodimentInfo] = Field(default_factory=list)

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 
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
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 
534
535
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 
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
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 
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
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 
 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
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 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
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'

JobStatusTransition

Bases: BaseModel

One entry in a job's status-change audit trail.

Emitted by the orchestrator's GET /jobs/{id}/status-transitions so users can trace a job's lifecycle. detail carries optional context (e.g. the cell id on dispatch/run, or the failure message on a terminal state).

Source code in core/src/armnet_core/models.py 
483
484
485
486
487
488
489
490
491
492
493
class JobStatusTransition(BaseModel):
    """One entry in a job's status-change audit trail.

    Emitted by the orchestrator's ``GET /jobs/{id}/status-transitions`` so users
    can trace a job's lifecycle. ``detail`` carries optional context (e.g. the
    cell id on dispatch/run, or the failure message on a terminal state).
    """

    status: str
    detail: Optional[str] = None
    timestamp: datetime

status instance-attribute 

status: str

detail class-attribute instance-attribute 

detail: Optional[str] = None

timestamp instance-attribute 

timestamp: datetime

RegistryCredentials

Bases: BaseModel

Short-lived credentials for pushing to the platform's image registry.

Returned by POST /registry/credentials. The orchestrator mints these on demand by impersonating a dedicated image-pusher service account. Customers never need to know about Google Cloud, IAM, or service accounts — they just Image.build(...).push().

The password is an OAuth2 access token valid for ~1 hour. The SDK caches it on disk and refreshes within 5 minutes of expiry.

Source code in core/src/armnet_core/models.py 
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
class RegistryCredentials(BaseModel):
    """Short-lived credentials for pushing to the platform's image registry.

    Returned by ``POST /registry/credentials``. The orchestrator mints
    these on demand by impersonating a dedicated image-pusher service
    account. Customers never need to know about Google Cloud, IAM, or
    service accounts \u2014 they just ``Image.build(...).push()``.

    The ``password`` is an OAuth2 access token valid for ~1 hour. The
    SDK caches it on disk and refreshes within 5 minutes of expiry.
    """

    registry: str = Field(
        ...,
        description="Registry hostname, e.g. 'europe-north1-docker.pkg.dev'.",
    )
    namespace: str = Field(
        ...,
        description="Path under the registry the customer is allowed to "
        "push to: '<project>/<repo>/<customer-id>'. The full image ref "
        "is '<registry>/<namespace>/<image-name>:<tag>'.",
    )
    username: str = Field(
        ...,
        description="docker login username. Always 'oauth2accesstoken' for AR.",
    )
    password: str = Field(
        ...,
        description="OAuth2 access token. Treat as a secret \u2014 valid for "
        "~1 hour and grants write access to the namespace above.",
    )
    expires_at: datetime = Field(
        ...,
        description="Wall-clock time at which the password becomes invalid.",
    )

registry class-attribute instance-attribute 

registry: str = Field(..., description="Registry hostname, e.g. 'europe-north1-docker.pkg.dev'.")

namespace class-attribute instance-attribute 

namespace: str = Field(..., description="Path under the registry the customer is allowed to push to: '<project>/<repo>/<customer-id>'. The full image ref is '<registry>/<namespace>/<image-name>:<tag>'.")

username class-attribute instance-attribute 

username: str = Field(..., description="docker login username. Always 'oauth2accesstoken' for AR.")

password class-attribute instance-attribute 

password: str = Field(..., description='OAuth2 access token. Treat as a secret — valid for ~1 hour and grants write access to the namespace above.')

expires_at class-attribute instance-attribute 

expires_at: datetime = Field(..., description='Wall-clock time at which the password becomes invalid.')

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 
496
497
498
499
500
501
502
503
504
505
506
507
508
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

RobotArmConfig

Bases: BaseModel

Named arm configuration for multi-arm (bimanual) cells.

The runtime derives each arm's connector endpoint from the cell-level robot_connector_endpoint plus the arm name; port is the physical serial device that arm is wired to on the edge Pi, used by the Fleet Agent to provision the (bimanual) edge connector with the per-arm port mapping (--arm-ports). Optional because the edge can also be provisioned out-of-band, but set it so the agent can bring a bimanual cell up itself.

Source code in core/src/armnet_core/models.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class RobotArmConfig(BaseModel):
    """Named arm configuration for multi-arm (bimanual) cells.

    The runtime derives each arm's connector endpoint from the cell-level
    ``robot_connector_endpoint`` plus the arm name; ``port`` is the physical
    serial device that arm is wired to on the edge Pi, used by the Fleet Agent
    to provision the (bimanual) edge connector with the per-arm port mapping
    (``--arm-ports``). Optional because the edge can also be provisioned
    out-of-band, but set it so the agent can bring a bimanual cell up itself.
    """

    model_config = ConfigDict(extra="forbid")

    robot_id: Optional[str] = None
    calibration_dir: Optional[Path] = None
    calibration_file_path: Optional[Path] = None
    rest_position: Optional[dict[str, float]] = None
    safety_limit: Optional[float] = None
    safety_delta_degrees: Optional[float] = None
    power_plug_ip: Optional[str] = None
    port: Optional[str] = None

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

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

rest_position class-attribute instance-attribute 

rest_position: Optional[dict[str, float]] = None

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = None

safety_delta_degrees class-attribute instance-attribute 

safety_delta_degrees: Optional[float] = None

power_plug_ip class-attribute instance-attribute 

power_plug_ip: Optional[str] = None

port class-attribute instance-attribute 

port: Optional[str] = None

RobotCellConfig

Bases: BaseModel

Physical robot cell configuration loaded from robot_cell.json.

This is operator-owned configuration, not job input. The cell and local Docker runner use it to inject the same robot metadata into the runtime context for every job that lands on this cell.

The JSON file may use a nested structure with top-level groups: cell, robot, data_interface, policy. A model validator flattens nested input into the canonical flat field set for backward compatibility with all existing consumers.

Source code in core/src/armnet_core/models.py 
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
class RobotCellConfig(BaseModel):
    """Physical robot cell configuration loaded from ``robot_cell.json``.

    This is operator-owned configuration, not job input. The cell and local
    Docker runner use it to inject the same robot metadata into the runtime
    context for every job that lands on this cell.

    The JSON file may use a nested structure with top-level groups:
    ``cell``, ``robot``, ``data_interface``, ``policy``. A model validator
    flattens nested input into the canonical flat field set for backward
    compatibility with all existing consumers.
    """

    nats_url: Optional[str] = None
    cell_id: Optional[str] = None
    embodiment: Embodiment = DEFAULT_EMBODIMENT
    task: Task = DEFAULT_TASK
    connector_socket_path: Optional[str] = None
    connector_tcp: Optional[str] = None
    robot_connector_endpoint: Optional[str] = None
    operator_call_endpoint: Optional[str] = "tcp://127.0.0.1:9877"
    robot_port: Optional[str] = None
    robot_power_plug_ip: Optional[str] = None
    robot_id: Optional[str] = None
    calibration_dir: Optional[Path] = None
    calibration_file_path: Optional[Path] = None
    camera_configs: dict[str, dict[str, Any]] = Field(default_factory=dict)
    arms: dict[str, RobotArmConfig] = Field(default_factory=dict)
    safety_limit: Optional[float] = Field(
        default=None,
        description=(
            "Runtime-facing relative action safety limit for robot SDK configs. "
            "SO-101 cells expose 30 degrees; embodiments with connector-only "
            "safety, such as ARX5, leave this unset."
        ),
    )
    docker_gpus: Optional[str] = None
    language_instruction: Optional[str] = None
    completion_enabled: bool = False
    completion_threshold: float = 0.9
    completion_camera: Optional[str] = None
    completion_min_interval_s: float = 1.0
    completion_model_path: str = "robometer/Robometer-4B"
    completion_chunk_seconds: float = 2.0
    completion_sample_fps: float = 4.0
    arm_rest_position: Optional[dict[str, float]] = None
    # Robot motion/safety params (pushed to edge connector at job start)
    safety_delta_degrees: float = 30.0
    rest_return_seconds: float = 4.0
    rest_return_hz: float = 20.0
    gripper_release_seconds: float = 1.0
    gripper_motor_name: str = "gripper"
    gripper_release_offset: float = 10.0
    # Data interface params
    tcp_read_timeout: float = 10.0
    jpeg_quality: int = 0
    # Cell infra
    shelly_timeout: float = 2.0

    @classmethod
    def _flatten_nested(cls, data: dict[str, Any]) -> dict[str, Any]:
        """Flatten nested JSON groups into the canonical flat field set."""
        flat: dict[str, Any] = {}

        cell = data.pop("cell", None)
        if isinstance(cell, dict):
            flat["nats_url"] = cell.get("nats_url")
            flat["cell_id"] = cell.get("cell_id")
            flat["connector_socket_path"] = cell.get("connector_socket_path")
            flat["connector_tcp"] = cell.get("connector_tcp")
            flat["robot_connector_endpoint"] = cell.get("robot_connector_endpoint")
            flat["operator_call_endpoint"] = cell.get("operator_call_endpoint")
            flat["robot_port"] = cell.get("robot_port")
            flat["docker_gpus"] = cell.get("docker_gpus")
            flat["robot_power_plug_ip"] = cell.get("power_plug_ip")
            flat["shelly_timeout"] = cell.get("shelly_timeout", 2.0)

        robot = data.pop("robot", None)
        if isinstance(robot, dict):
            flat["embodiment"] = robot.get("embodiment")
            flat["robot_id"] = robot.get("robot_id")
            flat["calibration_dir"] = robot.get("calibration_dir")
            flat["calibration_file_path"] = robot.get("calibration_file_path")
            flat["arm_rest_position"] = robot.get("rest_position")
            flat["camera_configs"] = robot.get("camera_configs", {})
            flat["arms"] = robot.get("arms", {})
            flat["safety_delta_degrees"] = robot.get("safety_delta_degrees", 30.0)
            flat["rest_return_seconds"] = robot.get("rest_return_seconds", 4.0)
            flat["rest_return_hz"] = robot.get("rest_return_hz", 20.0)
            flat["gripper_release_seconds"] = robot.get("gripper_release_seconds", 1.0)
            flat["gripper_motor_name"] = robot.get("gripper_motor_name", "gripper")
            flat["gripper_release_offset"] = robot.get("gripper_release_offset", 10.0)

        data_interface = data.pop("data_interface", None)
        if isinstance(data_interface, dict):
            flat["tcp_read_timeout"] = data_interface.get("tcp_read_timeout", 10.0)
            flat["jpeg_quality"] = data_interface.get("jpeg_quality", 0)

        policy = data.pop("policy", None)
        if isinstance(policy, dict):
            # NOTE: ``task`` / ``language_instruction`` are intentionally NOT read
            # from the config file. They are database-driven: the FMS owns the
            # cell's assigned task and forwards it (plus the task description) to
            # the cell via env vars (see cell/config.py). Any ``policy.task`` /
            # ``policy.language_instruction`` left in a JSON file is ignored so a
            # stale copy can't override the DB.
            completion = policy.get("completion", {})
            if isinstance(completion, dict):
                flat["completion_enabled"] = completion.get("enabled", False)
                flat["completion_threshold"] = completion.get("threshold", 0.9)
                flat["completion_camera"] = completion.get("camera")
                flat["completion_min_interval_s"] = completion.get("min_interval_s", 1.0)
                flat["completion_model_path"] = completion.get(
                    "model_path", "robometer/Robometer-4B"
                )
                flat["completion_chunk_seconds"] = completion.get("chunk_seconds", 2.0)
                flat["completion_sample_fps"] = completion.get("sample_fps", 4.0)

        # Remove None values so Pydantic defaults apply
        flat = {k: v for k, v in flat.items() if v is not None}
        # Merge: explicit flat fields in data take precedence over unpacked nested
        flat.update(data)
        return flat

    @model_validator(mode="before")
    @classmethod
    def _accept_nested_format(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        if any(key in data for key in ("cell", "robot", "policy", "data_interface")):
            return cls._flatten_nested(dict(data))
        return data

    @model_validator(mode="after")
    def _validate_named_arms(self) -> "RobotCellConfig":
        if self.arms and not {"left", "right"}.issubset(self.arms):
            raise ValueError("bimanual robot.arms must include both 'left' and 'right'")
        return self

    @classmethod
    def from_file(cls, path: str | Path) -> "RobotCellConfig":
        config_path = Path(path).expanduser().resolve()
        config = cls.model_validate_json(config_path.read_text())
        update: dict[str, Any] = {}
        if config.calibration_dir and not config.calibration_dir.is_absolute():
            update["calibration_dir"] = config_path.parent / config.calibration_dir
        if config.calibration_file_path and not config.calibration_file_path.is_absolute():
            update["calibration_file_path"] = config_path.parent / config.calibration_file_path
        if config.arms:
            arms: dict[str, RobotArmConfig] = {}
            changed = False
            for name, arm in config.arms.items():
                arm_update: dict[str, Path] = {}
                if arm.calibration_dir and not arm.calibration_dir.is_absolute():
                    arm_update["calibration_dir"] = config_path.parent / arm.calibration_dir
                if arm.calibration_file_path and not arm.calibration_file_path.is_absolute():
                    arm_update["calibration_file_path"] = (
                        config_path.parent / arm.calibration_file_path
                    )
                arms[name] = arm.model_copy(update=arm_update) if arm_update else arm
                changed = changed or bool(arm_update)
            if changed:
                update["arms"] = arms
        return config.model_copy(update=update) if update else config

nats_url class-attribute instance-attribute 

nats_url: Optional[str] = None

cell_id class-attribute instance-attribute 

cell_id: Optional[str] = None

embodiment class-attribute instance-attribute 

embodiment: Embodiment = DEFAULT_EMBODIMENT

task class-attribute instance-attribute 

task: Task = DEFAULT_TASK

connector_socket_path class-attribute instance-attribute 

connector_socket_path: Optional[str] = None

connector_tcp class-attribute instance-attribute 

connector_tcp: Optional[str] = None

robot_connector_endpoint class-attribute instance-attribute 

robot_connector_endpoint: Optional[str] = None

operator_call_endpoint class-attribute instance-attribute 

operator_call_endpoint: Optional[str] = 'tcp://127.0.0.1:9877'

robot_port class-attribute instance-attribute 

robot_port: Optional[str] = None

robot_power_plug_ip class-attribute instance-attribute 

robot_power_plug_ip: Optional[str] = None

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

camera_configs class-attribute instance-attribute 

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

arms class-attribute instance-attribute 

arms: dict[str, RobotArmConfig] = Field(default_factory=dict)

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = Field(default=None, description='Runtime-facing relative action safety limit for robot SDK configs. SO-101 cells expose 30 degrees; embodiments with connector-only safety, such as ARX5, leave this unset.')

docker_gpus class-attribute instance-attribute 

docker_gpus: Optional[str] = None

language_instruction class-attribute instance-attribute 

language_instruction: Optional[str] = None

completion_enabled class-attribute instance-attribute 

completion_enabled: bool = False

completion_threshold class-attribute instance-attribute 

completion_threshold: float = 0.9

completion_camera class-attribute instance-attribute 

completion_camera: Optional[str] = None

completion_min_interval_s class-attribute instance-attribute 

completion_min_interval_s: float = 1.0

completion_model_path class-attribute instance-attribute 

completion_model_path: str = 'robometer/Robometer-4B'

completion_chunk_seconds class-attribute instance-attribute 

completion_chunk_seconds: float = 2.0

completion_sample_fps class-attribute instance-attribute 

completion_sample_fps: float = 4.0

arm_rest_position class-attribute instance-attribute 

arm_rest_position: Optional[dict[str, float]] = None

safety_delta_degrees class-attribute instance-attribute 

safety_delta_degrees: float = 30.0

rest_return_seconds class-attribute instance-attribute 

rest_return_seconds: float = 4.0

rest_return_hz class-attribute instance-attribute 

rest_return_hz: float = 20.0

gripper_release_seconds class-attribute instance-attribute 

gripper_release_seconds: float = 1.0

gripper_motor_name class-attribute instance-attribute 

gripper_motor_name: str = 'gripper'

gripper_release_offset class-attribute instance-attribute 

gripper_release_offset: float = 10.0

tcp_read_timeout class-attribute instance-attribute 

tcp_read_timeout: float = 10.0

jpeg_quality class-attribute instance-attribute 

jpeg_quality: int = 0

shelly_timeout class-attribute instance-attribute 

shelly_timeout: float = 2.0

from_file classmethod 

from_file(path: str | Path) -> 'RobotCellConfig'
Source code in core/src/armnet_core/models.py 
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
@classmethod
def from_file(cls, path: str | Path) -> "RobotCellConfig":
    config_path = Path(path).expanduser().resolve()
    config = cls.model_validate_json(config_path.read_text())
    update: dict[str, Any] = {}
    if config.calibration_dir and not config.calibration_dir.is_absolute():
        update["calibration_dir"] = config_path.parent / config.calibration_dir
    if config.calibration_file_path and not config.calibration_file_path.is_absolute():
        update["calibration_file_path"] = config_path.parent / config.calibration_file_path
    if config.arms:
        arms: dict[str, RobotArmConfig] = {}
        changed = False
        for name, arm in config.arms.items():
            arm_update: dict[str, Path] = {}
            if arm.calibration_dir and not arm.calibration_dir.is_absolute():
                arm_update["calibration_dir"] = config_path.parent / arm.calibration_dir
            if arm.calibration_file_path and not arm.calibration_file_path.is_absolute():
                arm_update["calibration_file_path"] = (
                    config_path.parent / arm.calibration_file_path
                )
            arms[name] = arm.model_copy(update=arm_update) if arm_update else arm
            changed = changed or bool(arm_update)
        if changed:
            update["arms"] = arms
    return config.model_copy(update=update) if update else config

SecretCreateRequest

Bases: BaseModel

Create or replace one user secret.

Source code in core/src/armnet_core/models.py 
628
629
630
631
632
class SecretCreateRequest(BaseModel):
    """Create or replace one user secret."""

    name: str
    value: str

name instance-attribute 

name: str

value instance-attribute 

value: str

SecretInfo

Bases: BaseModel

Secret metadata returned to clients. Never includes secret values.

Source code in core/src/armnet_core/models.py 
616
617
618
619
class SecretInfo(BaseModel):
    """Secret metadata returned to clients. Never includes secret values."""

    name: str

name instance-attribute 

name: str

SecretList

Bases: BaseModel

List of secret metadata for the authenticated user.

Source code in core/src/armnet_core/models.py 
622
623
624
625
class SecretList(BaseModel):
    """List of secret metadata for the authenticated user."""

    secrets: list[SecretInfo] = Field(default_factory=list)

secrets class-attribute instance-attribute 

secrets: list[SecretInfo] = Field(default_factory=list)

TaskInfo

Bases: BaseModel

One known task, as stored in the orchestrator database.

Source code in core/src/armnet_core/models.py 
603
604
605
606
607
class TaskInfo(BaseModel):
    """One known task, as stored in the orchestrator database."""

    slug: Task = Field(..., description="Wire-format task slug, e.g. 'assemble_block_tower'.")
    description: Optional[str] = None

slug class-attribute instance-attribute 

slug: Task = Field(..., description="Wire-format task slug, e.g. 'assemble_block_tower'.")

description class-attribute instance-attribute 

description: Optional[str] = None

TaskList

Bases: BaseModel

The set of tasks the platform currently accepts (GET /tasks).

Source code in core/src/armnet_core/models.py 
610
611
612
613
class TaskList(BaseModel):
    """The set of tasks the platform currently accepts (GET /tasks)."""

    tasks: list[TaskInfo] = Field(default_factory=list)

tasks class-attribute instance-attribute 

tasks: list[TaskInfo] = Field(default_factory=list)

TeleopAction

Bases: BaseModel

A single remote-teleoperation action for a running job.

Sent by the client (sampling a local leader arm) to the orchestrator over a websocket, forwarded to the cell on the job's teleop NATS subject, and kept by the cell as a most-recent-value register. timestamp (unix seconds, on the client clock) is used to drop out-of-order/stale messages so the cell always holds the freshest action.

Source code in core/src/armnet_core/models.py 
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
class TeleopAction(BaseModel):
    """A single remote-teleoperation action for a running job.

    Sent by the client (sampling a local leader arm) to the orchestrator over a
    websocket, forwarded to the cell on the job's teleop NATS subject, and kept
    by the cell as a most-recent-value register. ``timestamp`` (unix seconds, on
    the client clock) is used to drop out-of-order/stale messages so the cell
    always holds the freshest action.
    """

    job_id: str
    timestamp: float
    # Joint-space command keyed like LeRobot's send_action input, e.g.
    # {"shoulder_pan.pos": 12.3, ...}.
    action: dict[str, float] = Field(default_factory=dict)

job_id instance-attribute 

job_id: str

timestamp instance-attribute 

timestamp: float

action class-attribute instance-attribute 

action: dict[str, float] = Field(default_factory=dict)

TeleopEvent

Bases: BaseModel

A discrete recording-control event for a running teleop job.

Travels the same path as :class:TeleopAction (client websocket → orchestrator → the job's teleop NATS subject → cell), but with different delivery semantics: actions are a freshest-wins stream where drops are fine, whereas events are rare, user-initiated commands that the cell queues in order so none is lost between runtime polls.

The event values mirror LeRobot's dataset-recording keyboard controls: Right Arrow ends the current episode and moves on (next_episode), Left Arrow discards and re-records it (rerecord_episode), and Esc stops the whole session (stop_recording).

Source code in core/src/armnet_core/models.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
class TeleopEvent(BaseModel):
    """A discrete recording-control event for a running teleop job.

    Travels the same path as :class:`TeleopAction` (client websocket →
    orchestrator → the job's teleop NATS subject → cell), but with different
    delivery semantics: actions are a freshest-wins stream where drops are
    fine, whereas events are rare, user-initiated commands that the cell queues
    in order so none is lost between runtime polls.

    The ``event`` values mirror LeRobot's dataset-recording keyboard controls:
    Right Arrow ends the current episode and moves on (``next_episode``), Left
    Arrow discards and re-records it (``rerecord_episode``), and Esc stops the
    whole session (``stop_recording``).
    """

    job_id: str
    timestamp: float
    event: str = Field(..., pattern="^(next_episode|rerecord_episode|stop_recording)$")

job_id instance-attribute 

job_id: str

timestamp instance-attribute 

timestamp: float

event class-attribute instance-attribute 

event: str = Field(..., pattern='^(next_episode|rerecord_episode|stop_recording)$')

VolumeCredentials

Bases: BaseModel

Short-lived credentials for a user's cloud-backed volume prefix.

Source code in core/src/armnet_core/models.py 
575
576
577
578
579
580
581
class VolumeCredentials(BaseModel):
    """Short-lived credentials for a user's cloud-backed volume prefix."""

    bucket: str
    prefix: str
    access_token: str
    expires_at: datetime

bucket instance-attribute 

bucket: str

prefix instance-attribute 

prefix: str

access_token instance-attribute 

access_token: str

expires_at instance-attribute 

expires_at: datetime

WhoAmI

Bases: BaseModel

Authenticated API identity.

Source code in core/src/armnet_core/models.py 
584
585
586
587
class WhoAmI(BaseModel):
    """Authenticated API identity."""

    username: str

username instance-attribute 

username: str

armnet_core.enums

Embodiment and Task wire-type aliases.

Embodiments and tasks are identified by stable wire-format string slugs that appear in HTTP request bodies, NATS subjects, env vars injected into customer containers, and cell configuration (e.g. "lerobot/so-101", "assemble_block_tower").

The set of valid embodiments and tasks is not hard-coded here. It lives in the orchestrator's Postgres database (the embodiments and tasks tables; see db/), which is the single source of truth. The orchestrator validates incoming jobs against those tables and rejects unknown values with HTTP 400. Adding a new embodiment or task is a database insert, not a code change.

Embodiment and Task are kept as str aliases so annotations stay readable and existing Embodiment(value)/Task(value) call sites keep working (they are now just str(value)). The DEFAULT_* constants below are convenience defaults for local development and SDK ergonomics only — they are not an authoritative enumeration of valid values.

Embodiment module-attribute 

Embodiment = str

Task module-attribute 

Task = str

DEFAULT_EMBODIMENT module-attribute 

DEFAULT_EMBODIMENT: Embodiment = 'lerobot/so-101'

DEFAULT_TASK module-attribute 

DEFAULT_TASK: Task = 'assemble_block_tower'

armnet_core.models

External-facing wire types for the armnet platform.

Customer-touching models only — request/response bodies for the orchestrator HTTP API and the result payloads embedded in those responses. Internal NATS message envelopes (JobDispatch, CellHeartbeat) live in the separate armnet-protocol package alongside the NATS subject taxonomy.

Keep this module dependency-light (pydantic + stdlib only) so it can be imported from any component without dragging in HTTP or NATS clients.

TerminalStatus module-attribute 

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

TELEOP_EVENT_NEXT_EPISODE module-attribute 

TELEOP_EVENT_NEXT_EPISODE = 'next_episode'

TELEOP_EVENT_RERECORD_EPISODE module-attribute 

TELEOP_EVENT_RERECORD_EPISODE = 'rerecord_episode'

TELEOP_EVENT_STOP_RECORDING module-attribute 

TELEOP_EVENT_STOP_RECORDING = 'stop_recording'

TELEOP_EVENTS module-attribute 

TELEOP_EVENTS = (TELEOP_EVENT_NEXT_EPISODE, TELEOP_EVENT_RERECORD_EPISODE, TELEOP_EVENT_STOP_RECORDING)

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 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
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'

CellRuntimeStatus

Bases: str, Enum

Current availability state for a robot cell.

Source code in core/src/armnet_core/models.py 
56
57
58
59
60
61
class CellRuntimeStatus(str, Enum):
    """Current availability state for a robot cell."""

    AVAILABLE = "available"
    OCCUPIED = "occupied"
    OFFLINE = "offline"

AVAILABLE class-attribute instance-attribute 

AVAILABLE = 'available'

OCCUPIED class-attribute instance-attribute 

OCCUPIED = 'occupied'

OFFLINE class-attribute instance-attribute 

OFFLINE = 'offline'

RobotArmConfig

Bases: BaseModel

Named arm configuration for multi-arm (bimanual) cells.

The runtime derives each arm's connector endpoint from the cell-level robot_connector_endpoint plus the arm name; port is the physical serial device that arm is wired to on the edge Pi, used by the Fleet Agent to provision the (bimanual) edge connector with the per-arm port mapping (--arm-ports). Optional because the edge can also be provisioned out-of-band, but set it so the agent can bring a bimanual cell up itself.

Source code in core/src/armnet_core/models.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class RobotArmConfig(BaseModel):
    """Named arm configuration for multi-arm (bimanual) cells.

    The runtime derives each arm's connector endpoint from the cell-level
    ``robot_connector_endpoint`` plus the arm name; ``port`` is the physical
    serial device that arm is wired to on the edge Pi, used by the Fleet Agent
    to provision the (bimanual) edge connector with the per-arm port mapping
    (``--arm-ports``). Optional because the edge can also be provisioned
    out-of-band, but set it so the agent can bring a bimanual cell up itself.
    """

    model_config = ConfigDict(extra="forbid")

    robot_id: Optional[str] = None
    calibration_dir: Optional[Path] = None
    calibration_file_path: Optional[Path] = None
    rest_position: Optional[dict[str, float]] = None
    safety_limit: Optional[float] = None
    safety_delta_degrees: Optional[float] = None
    power_plug_ip: Optional[str] = None
    port: Optional[str] = None

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

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

rest_position class-attribute instance-attribute 

rest_position: Optional[dict[str, float]] = None

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = None

safety_delta_degrees class-attribute instance-attribute 

safety_delta_degrees: Optional[float] = None

power_plug_ip class-attribute instance-attribute 

power_plug_ip: Optional[str] = None

port class-attribute instance-attribute 

port: Optional[str] = None

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

RobotCellConfig

Bases: BaseModel

Physical robot cell configuration loaded from robot_cell.json.

This is operator-owned configuration, not job input. The cell and local Docker runner use it to inject the same robot metadata into the runtime context for every job that lands on this cell.

The JSON file may use a nested structure with top-level groups: cell, robot, data_interface, policy. A model validator flattens nested input into the canonical flat field set for backward compatibility with all existing consumers.

Source code in core/src/armnet_core/models.py 
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
class RobotCellConfig(BaseModel):
    """Physical robot cell configuration loaded from ``robot_cell.json``.

    This is operator-owned configuration, not job input. The cell and local
    Docker runner use it to inject the same robot metadata into the runtime
    context for every job that lands on this cell.

    The JSON file may use a nested structure with top-level groups:
    ``cell``, ``robot``, ``data_interface``, ``policy``. A model validator
    flattens nested input into the canonical flat field set for backward
    compatibility with all existing consumers.
    """

    nats_url: Optional[str] = None
    cell_id: Optional[str] = None
    embodiment: Embodiment = DEFAULT_EMBODIMENT
    task: Task = DEFAULT_TASK
    connector_socket_path: Optional[str] = None
    connector_tcp: Optional[str] = None
    robot_connector_endpoint: Optional[str] = None
    operator_call_endpoint: Optional[str] = "tcp://127.0.0.1:9877"
    robot_port: Optional[str] = None
    robot_power_plug_ip: Optional[str] = None
    robot_id: Optional[str] = None
    calibration_dir: Optional[Path] = None
    calibration_file_path: Optional[Path] = None
    camera_configs: dict[str, dict[str, Any]] = Field(default_factory=dict)
    arms: dict[str, RobotArmConfig] = Field(default_factory=dict)
    safety_limit: Optional[float] = Field(
        default=None,
        description=(
            "Runtime-facing relative action safety limit for robot SDK configs. "
            "SO-101 cells expose 30 degrees; embodiments with connector-only "
            "safety, such as ARX5, leave this unset."
        ),
    )
    docker_gpus: Optional[str] = None
    language_instruction: Optional[str] = None
    completion_enabled: bool = False
    completion_threshold: float = 0.9
    completion_camera: Optional[str] = None
    completion_min_interval_s: float = 1.0
    completion_model_path: str = "robometer/Robometer-4B"
    completion_chunk_seconds: float = 2.0
    completion_sample_fps: float = 4.0
    arm_rest_position: Optional[dict[str, float]] = None
    # Robot motion/safety params (pushed to edge connector at job start)
    safety_delta_degrees: float = 30.0
    rest_return_seconds: float = 4.0
    rest_return_hz: float = 20.0
    gripper_release_seconds: float = 1.0
    gripper_motor_name: str = "gripper"
    gripper_release_offset: float = 10.0
    # Data interface params
    tcp_read_timeout: float = 10.0
    jpeg_quality: int = 0
    # Cell infra
    shelly_timeout: float = 2.0

    @classmethod
    def _flatten_nested(cls, data: dict[str, Any]) -> dict[str, Any]:
        """Flatten nested JSON groups into the canonical flat field set."""
        flat: dict[str, Any] = {}

        cell = data.pop("cell", None)
        if isinstance(cell, dict):
            flat["nats_url"] = cell.get("nats_url")
            flat["cell_id"] = cell.get("cell_id")
            flat["connector_socket_path"] = cell.get("connector_socket_path")
            flat["connector_tcp"] = cell.get("connector_tcp")
            flat["robot_connector_endpoint"] = cell.get("robot_connector_endpoint")
            flat["operator_call_endpoint"] = cell.get("operator_call_endpoint")
            flat["robot_port"] = cell.get("robot_port")
            flat["docker_gpus"] = cell.get("docker_gpus")
            flat["robot_power_plug_ip"] = cell.get("power_plug_ip")
            flat["shelly_timeout"] = cell.get("shelly_timeout", 2.0)

        robot = data.pop("robot", None)
        if isinstance(robot, dict):
            flat["embodiment"] = robot.get("embodiment")
            flat["robot_id"] = robot.get("robot_id")
            flat["calibration_dir"] = robot.get("calibration_dir")
            flat["calibration_file_path"] = robot.get("calibration_file_path")
            flat["arm_rest_position"] = robot.get("rest_position")
            flat["camera_configs"] = robot.get("camera_configs", {})
            flat["arms"] = robot.get("arms", {})
            flat["safety_delta_degrees"] = robot.get("safety_delta_degrees", 30.0)
            flat["rest_return_seconds"] = robot.get("rest_return_seconds", 4.0)
            flat["rest_return_hz"] = robot.get("rest_return_hz", 20.0)
            flat["gripper_release_seconds"] = robot.get("gripper_release_seconds", 1.0)
            flat["gripper_motor_name"] = robot.get("gripper_motor_name", "gripper")
            flat["gripper_release_offset"] = robot.get("gripper_release_offset", 10.0)

        data_interface = data.pop("data_interface", None)
        if isinstance(data_interface, dict):
            flat["tcp_read_timeout"] = data_interface.get("tcp_read_timeout", 10.0)
            flat["jpeg_quality"] = data_interface.get("jpeg_quality", 0)

        policy = data.pop("policy", None)
        if isinstance(policy, dict):
            # NOTE: ``task`` / ``language_instruction`` are intentionally NOT read
            # from the config file. They are database-driven: the FMS owns the
            # cell's assigned task and forwards it (plus the task description) to
            # the cell via env vars (see cell/config.py). Any ``policy.task`` /
            # ``policy.language_instruction`` left in a JSON file is ignored so a
            # stale copy can't override the DB.
            completion = policy.get("completion", {})
            if isinstance(completion, dict):
                flat["completion_enabled"] = completion.get("enabled", False)
                flat["completion_threshold"] = completion.get("threshold", 0.9)
                flat["completion_camera"] = completion.get("camera")
                flat["completion_min_interval_s"] = completion.get("min_interval_s", 1.0)
                flat["completion_model_path"] = completion.get(
                    "model_path", "robometer/Robometer-4B"
                )
                flat["completion_chunk_seconds"] = completion.get("chunk_seconds", 2.0)
                flat["completion_sample_fps"] = completion.get("sample_fps", 4.0)

        # Remove None values so Pydantic defaults apply
        flat = {k: v for k, v in flat.items() if v is not None}
        # Merge: explicit flat fields in data take precedence over unpacked nested
        flat.update(data)
        return flat

    @model_validator(mode="before")
    @classmethod
    def _accept_nested_format(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        if any(key in data for key in ("cell", "robot", "policy", "data_interface")):
            return cls._flatten_nested(dict(data))
        return data

    @model_validator(mode="after")
    def _validate_named_arms(self) -> "RobotCellConfig":
        if self.arms and not {"left", "right"}.issubset(self.arms):
            raise ValueError("bimanual robot.arms must include both 'left' and 'right'")
        return self

    @classmethod
    def from_file(cls, path: str | Path) -> "RobotCellConfig":
        config_path = Path(path).expanduser().resolve()
        config = cls.model_validate_json(config_path.read_text())
        update: dict[str, Any] = {}
        if config.calibration_dir and not config.calibration_dir.is_absolute():
            update["calibration_dir"] = config_path.parent / config.calibration_dir
        if config.calibration_file_path and not config.calibration_file_path.is_absolute():
            update["calibration_file_path"] = config_path.parent / config.calibration_file_path
        if config.arms:
            arms: dict[str, RobotArmConfig] = {}
            changed = False
            for name, arm in config.arms.items():
                arm_update: dict[str, Path] = {}
                if arm.calibration_dir and not arm.calibration_dir.is_absolute():
                    arm_update["calibration_dir"] = config_path.parent / arm.calibration_dir
                if arm.calibration_file_path and not arm.calibration_file_path.is_absolute():
                    arm_update["calibration_file_path"] = (
                        config_path.parent / arm.calibration_file_path
                    )
                arms[name] = arm.model_copy(update=arm_update) if arm_update else arm
                changed = changed or bool(arm_update)
            if changed:
                update["arms"] = arms
        return config.model_copy(update=update) if update else config

nats_url class-attribute instance-attribute 

nats_url: Optional[str] = None

cell_id class-attribute instance-attribute 

cell_id: Optional[str] = None

embodiment class-attribute instance-attribute 

embodiment: Embodiment = DEFAULT_EMBODIMENT

task class-attribute instance-attribute 

task: Task = DEFAULT_TASK

connector_socket_path class-attribute instance-attribute 

connector_socket_path: Optional[str] = None

connector_tcp class-attribute instance-attribute 

connector_tcp: Optional[str] = None

robot_connector_endpoint class-attribute instance-attribute 

robot_connector_endpoint: Optional[str] = None

operator_call_endpoint class-attribute instance-attribute 

operator_call_endpoint: Optional[str] = 'tcp://127.0.0.1:9877'

robot_port class-attribute instance-attribute 

robot_port: Optional[str] = None

robot_power_plug_ip class-attribute instance-attribute 

robot_power_plug_ip: Optional[str] = None

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

camera_configs class-attribute instance-attribute 

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

arms class-attribute instance-attribute 

arms: dict[str, RobotArmConfig] = Field(default_factory=dict)

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = Field(default=None, description='Runtime-facing relative action safety limit for robot SDK configs. SO-101 cells expose 30 degrees; embodiments with connector-only safety, such as ARX5, leave this unset.')

docker_gpus class-attribute instance-attribute 

docker_gpus: Optional[str] = None

language_instruction class-attribute instance-attribute 

language_instruction: Optional[str] = None

completion_enabled class-attribute instance-attribute 

completion_enabled: bool = False

completion_threshold class-attribute instance-attribute 

completion_threshold: float = 0.9

completion_camera class-attribute instance-attribute 

completion_camera: Optional[str] = None

completion_min_interval_s class-attribute instance-attribute 

completion_min_interval_s: float = 1.0

completion_model_path class-attribute instance-attribute 

completion_model_path: str = 'robometer/Robometer-4B'

completion_chunk_seconds class-attribute instance-attribute 

completion_chunk_seconds: float = 2.0

completion_sample_fps class-attribute instance-attribute 

completion_sample_fps: float = 4.0

arm_rest_position class-attribute instance-attribute 

arm_rest_position: Optional[dict[str, float]] = None

safety_delta_degrees class-attribute instance-attribute 

safety_delta_degrees: float = 30.0

rest_return_seconds class-attribute instance-attribute 

rest_return_seconds: float = 4.0

rest_return_hz class-attribute instance-attribute 

rest_return_hz: float = 20.0

gripper_release_seconds class-attribute instance-attribute 

gripper_release_seconds: float = 1.0

gripper_motor_name class-attribute instance-attribute 

gripper_motor_name: str = 'gripper'

gripper_release_offset class-attribute instance-attribute 

gripper_release_offset: float = 10.0

tcp_read_timeout class-attribute instance-attribute 

tcp_read_timeout: float = 10.0

jpeg_quality class-attribute instance-attribute 

jpeg_quality: int = 0

shelly_timeout class-attribute instance-attribute 

shelly_timeout: float = 2.0

from_file classmethod 

from_file(path: str | Path) -> 'RobotCellConfig'
Source code in core/src/armnet_core/models.py 
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
@classmethod
def from_file(cls, path: str | Path) -> "RobotCellConfig":
    config_path = Path(path).expanduser().resolve()
    config = cls.model_validate_json(config_path.read_text())
    update: dict[str, Any] = {}
    if config.calibration_dir and not config.calibration_dir.is_absolute():
        update["calibration_dir"] = config_path.parent / config.calibration_dir
    if config.calibration_file_path and not config.calibration_file_path.is_absolute():
        update["calibration_file_path"] = config_path.parent / config.calibration_file_path
    if config.arms:
        arms: dict[str, RobotArmConfig] = {}
        changed = False
        for name, arm in config.arms.items():
            arm_update: dict[str, Path] = {}
            if arm.calibration_dir and not arm.calibration_dir.is_absolute():
                arm_update["calibration_dir"] = config_path.parent / arm.calibration_dir
            if arm.calibration_file_path and not arm.calibration_file_path.is_absolute():
                arm_update["calibration_file_path"] = (
                    config_path.parent / arm.calibration_file_path
                )
            arms[name] = arm.model_copy(update=arm_update) if arm_update else arm
            changed = changed or bool(arm_update)
        if changed:
            update["arms"] = arms
    return config.model_copy(update=update) if update else config

CellStatus

Bases: BaseModel

Current status for one cell as reported by recent heartbeats.

Source code in core/src/armnet_core/models.py 
323
324
325
326
327
328
329
330
331
class CellStatus(BaseModel):
    """Current status for one cell as reported by recent heartbeats."""

    cell_id: str
    embodiment: Embodiment
    task: Task
    status: CellRuntimeStatus
    current_job_id: Optional[str] = None
    last_heartbeat_at: Optional[datetime] = None

cell_id instance-attribute 

cell_id: str

embodiment instance-attribute 

embodiment: Embodiment

task instance-attribute 

task: Task

status instance-attribute 

status: CellRuntimeStatus

current_job_id class-attribute instance-attribute 

current_job_id: Optional[str] = None

last_heartbeat_at class-attribute instance-attribute 

last_heartbeat_at: Optional[datetime] = None

CellStatusSummary

Bases: BaseModel

Aggregate availability for an embodiment, optionally scoped to a task.

task is None when the summary aggregates across every task of the embodiment (used to answer "is any cell of this embodiment available?" for task-less jobs).

Source code in core/src/armnet_core/models.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
class CellStatusSummary(BaseModel):
    """Aggregate availability for an embodiment, optionally scoped to a task.

    ``task`` is ``None`` when the summary aggregates across every task of the
    embodiment (used to answer "is any cell of this embodiment available?" for
    task-less jobs).
    """

    embodiment: Embodiment
    task: Optional[Task] = None
    status: CellRuntimeStatus
    cells: list[CellStatus] = Field(default_factory=list)
    heartbeat_timeout_seconds: int = 20

embodiment instance-attribute 

embodiment: Embodiment

task class-attribute instance-attribute 

task: Optional[Task] = None

status instance-attribute 

status: CellRuntimeStatus

cells class-attribute instance-attribute 

cells: list[CellStatus] = Field(default_factory=list)

heartbeat_timeout_seconds class-attribute instance-attribute 

heartbeat_timeout_seconds: int = 20

TeleopAction

Bases: BaseModel

A single remote-teleoperation action for a running job.

Sent by the client (sampling a local leader arm) to the orchestrator over a websocket, forwarded to the cell on the job's teleop NATS subject, and kept by the cell as a most-recent-value register. timestamp (unix seconds, on the client clock) is used to drop out-of-order/stale messages so the cell always holds the freshest action.

Source code in core/src/armnet_core/models.py 
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
class TeleopAction(BaseModel):
    """A single remote-teleoperation action for a running job.

    Sent by the client (sampling a local leader arm) to the orchestrator over a
    websocket, forwarded to the cell on the job's teleop NATS subject, and kept
    by the cell as a most-recent-value register. ``timestamp`` (unix seconds, on
    the client clock) is used to drop out-of-order/stale messages so the cell
    always holds the freshest action.
    """

    job_id: str
    timestamp: float
    # Joint-space command keyed like LeRobot's send_action input, e.g.
    # {"shoulder_pan.pos": 12.3, ...}.
    action: dict[str, float] = Field(default_factory=dict)

job_id instance-attribute 

job_id: str

timestamp instance-attribute 

timestamp: float

action class-attribute instance-attribute 

action: dict[str, float] = Field(default_factory=dict)

TeleopEvent

Bases: BaseModel

A discrete recording-control event for a running teleop job.

Travels the same path as :class:TeleopAction (client websocket → orchestrator → the job's teleop NATS subject → cell), but with different delivery semantics: actions are a freshest-wins stream where drops are fine, whereas events are rare, user-initiated commands that the cell queues in order so none is lost between runtime polls.

The event values mirror LeRobot's dataset-recording keyboard controls: Right Arrow ends the current episode and moves on (next_episode), Left Arrow discards and re-records it (rerecord_episode), and Esc stops the whole session (stop_recording).

Source code in core/src/armnet_core/models.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
class TeleopEvent(BaseModel):
    """A discrete recording-control event for a running teleop job.

    Travels the same path as :class:`TeleopAction` (client websocket →
    orchestrator → the job's teleop NATS subject → cell), but with different
    delivery semantics: actions are a freshest-wins stream where drops are
    fine, whereas events are rare, user-initiated commands that the cell queues
    in order so none is lost between runtime polls.

    The ``event`` values mirror LeRobot's dataset-recording keyboard controls:
    Right Arrow ends the current episode and moves on (``next_episode``), Left
    Arrow discards and re-records it (``rerecord_episode``), and Esc stops the
    whole session (``stop_recording``).
    """

    job_id: str
    timestamp: float
    event: str = Field(..., pattern="^(next_episode|rerecord_episode|stop_recording)$")

job_id instance-attribute 

job_id: str

timestamp instance-attribute 

timestamp: float

event class-attribute instance-attribute 

event: str = Field(..., pattern='^(next_episode|rerecord_episode|stop_recording)$')

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 
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
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 
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
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)

JobStatusTransition

Bases: BaseModel

One entry in a job's status-change audit trail.

Emitted by the orchestrator's GET /jobs/{id}/status-transitions so users can trace a job's lifecycle. detail carries optional context (e.g. the cell id on dispatch/run, or the failure message on a terminal state).

Source code in core/src/armnet_core/models.py 
483
484
485
486
487
488
489
490
491
492
493
class JobStatusTransition(BaseModel):
    """One entry in a job's status-change audit trail.

    Emitted by the orchestrator's ``GET /jobs/{id}/status-transitions`` so users
    can trace a job's lifecycle. ``detail`` carries optional context (e.g. the
    cell id on dispatch/run, or the failure message on a terminal state).
    """

    status: str
    detail: Optional[str] = None
    timestamp: datetime

status instance-attribute 

status: str

detail class-attribute instance-attribute 

detail: Optional[str] = None

timestamp instance-attribute 

timestamp: datetime

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 
496
497
498
499
500
501
502
503
504
505
506
507
508
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

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 
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
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 
534
535
def is_terminal(self) -> bool:
    return self.status in TerminalStatus

RegistryCredentials

Bases: BaseModel

Short-lived credentials for pushing to the platform's image registry.

Returned by POST /registry/credentials. The orchestrator mints these on demand by impersonating a dedicated image-pusher service account. Customers never need to know about Google Cloud, IAM, or service accounts — they just Image.build(...).push().

The password is an OAuth2 access token valid for ~1 hour. The SDK caches it on disk and refreshes within 5 minutes of expiry.

Source code in core/src/armnet_core/models.py 
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
class RegistryCredentials(BaseModel):
    """Short-lived credentials for pushing to the platform's image registry.

    Returned by ``POST /registry/credentials``. The orchestrator mints
    these on demand by impersonating a dedicated image-pusher service
    account. Customers never need to know about Google Cloud, IAM, or
    service accounts \u2014 they just ``Image.build(...).push()``.

    The ``password`` is an OAuth2 access token valid for ~1 hour. The
    SDK caches it on disk and refreshes within 5 minutes of expiry.
    """

    registry: str = Field(
        ...,
        description="Registry hostname, e.g. 'europe-north1-docker.pkg.dev'.",
    )
    namespace: str = Field(
        ...,
        description="Path under the registry the customer is allowed to "
        "push to: '<project>/<repo>/<customer-id>'. The full image ref "
        "is '<registry>/<namespace>/<image-name>:<tag>'.",
    )
    username: str = Field(
        ...,
        description="docker login username. Always 'oauth2accesstoken' for AR.",
    )
    password: str = Field(
        ...,
        description="OAuth2 access token. Treat as a secret \u2014 valid for "
        "~1 hour and grants write access to the namespace above.",
    )
    expires_at: datetime = Field(
        ...,
        description="Wall-clock time at which the password becomes invalid.",
    )

registry class-attribute instance-attribute 

registry: str = Field(..., description="Registry hostname, e.g. 'europe-north1-docker.pkg.dev'.")

namespace class-attribute instance-attribute 

namespace: str = Field(..., description="Path under the registry the customer is allowed to push to: '<project>/<repo>/<customer-id>'. The full image ref is '<registry>/<namespace>/<image-name>:<tag>'.")

username class-attribute instance-attribute 

username: str = Field(..., description="docker login username. Always 'oauth2accesstoken' for AR.")

password class-attribute instance-attribute 

password: str = Field(..., description='OAuth2 access token. Treat as a secret — valid for ~1 hour and grants write access to the namespace above.')

expires_at class-attribute instance-attribute 

expires_at: datetime = Field(..., description='Wall-clock time at which the password becomes invalid.')

VolumeCredentials

Bases: BaseModel

Short-lived credentials for a user's cloud-backed volume prefix.

Source code in core/src/armnet_core/models.py 
575
576
577
578
579
580
581
class VolumeCredentials(BaseModel):
    """Short-lived credentials for a user's cloud-backed volume prefix."""

    bucket: str
    prefix: str
    access_token: str
    expires_at: datetime

bucket instance-attribute 

bucket: str

prefix instance-attribute 

prefix: str

access_token instance-attribute 

access_token: str

expires_at instance-attribute 

expires_at: datetime

WhoAmI

Bases: BaseModel

Authenticated API identity.

Source code in core/src/armnet_core/models.py 
584
585
586
587
class WhoAmI(BaseModel):
    """Authenticated API identity."""

    username: str

username instance-attribute 

username: str

EmbodimentInfo

Bases: BaseModel

One known embodiment, as stored in the orchestrator database.

Source code in core/src/armnet_core/models.py 
590
591
592
593
594
class EmbodimentInfo(BaseModel):
    """One known embodiment, as stored in the orchestrator database."""

    slug: Embodiment = Field(..., description="Wire-format embodiment slug, e.g. 'lerobot/so-101'.")
    description: Optional[str] = None

slug class-attribute instance-attribute 

slug: Embodiment = Field(..., description="Wire-format embodiment slug, e.g. 'lerobot/so-101'.")

description class-attribute instance-attribute 

description: Optional[str] = None

EmbodimentList

Bases: BaseModel

The set of embodiments the platform currently accepts (GET /embodiments).

Source code in core/src/armnet_core/models.py 
597
598
599
600
class EmbodimentList(BaseModel):
    """The set of embodiments the platform currently accepts (GET /embodiments)."""

    embodiments: list[EmbodimentInfo] = Field(default_factory=list)

embodiments class-attribute instance-attribute 

embodiments: list[EmbodimentInfo] = Field(default_factory=list)

TaskInfo

Bases: BaseModel

One known task, as stored in the orchestrator database.

Source code in core/src/armnet_core/models.py 
603
604
605
606
607
class TaskInfo(BaseModel):
    """One known task, as stored in the orchestrator database."""

    slug: Task = Field(..., description="Wire-format task slug, e.g. 'assemble_block_tower'.")
    description: Optional[str] = None

slug class-attribute instance-attribute 

slug: Task = Field(..., description="Wire-format task slug, e.g. 'assemble_block_tower'.")

description class-attribute instance-attribute 

description: Optional[str] = None

TaskList

Bases: BaseModel

The set of tasks the platform currently accepts (GET /tasks).

Source code in core/src/armnet_core/models.py 
610
611
612
613
class TaskList(BaseModel):
    """The set of tasks the platform currently accepts (GET /tasks)."""

    tasks: list[TaskInfo] = Field(default_factory=list)

tasks class-attribute instance-attribute 

tasks: list[TaskInfo] = Field(default_factory=list)

SecretInfo

Bases: BaseModel

Secret metadata returned to clients. Never includes secret values.

Source code in core/src/armnet_core/models.py 
616
617
618
619
class SecretInfo(BaseModel):
    """Secret metadata returned to clients. Never includes secret values."""

    name: str

name instance-attribute 

name: str

SecretList

Bases: BaseModel

List of secret metadata for the authenticated user.

Source code in core/src/armnet_core/models.py 
622
623
624
625
class SecretList(BaseModel):
    """List of secret metadata for the authenticated user."""

    secrets: list[SecretInfo] = Field(default_factory=list)

secrets class-attribute instance-attribute 

secrets: list[SecretInfo] = Field(default_factory=list)

SecretCreateRequest

Bases: BaseModel

Create or replace one user secret.

Source code in core/src/armnet_core/models.py 
628
629
630
631
632
class SecretCreateRequest(BaseModel):
    """Create or replace one user secret."""

    name: str
    value: str

name instance-attribute 

name: str

value instance-attribute 

value: str

armnet_core.auth

Shared authentication constants.

API_KEY_ENV module-attribute 

API_KEY_ENV = 'ARMNET_API_KEY'

API_KEY_HEADER module-attribute 

API_KEY_HEADER = 'X-Armnet-Api-Key'