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_ENVIRONMENT module-attribute 

DEFAULT_ENVIRONMENT: Environment = 'busybox'

DEFAULT_TASK module-attribute 

DEFAULT_TASK: Task = 'assemble_block_tower'

Embodiment module-attribute 

Embodiment = str

EnvironmentName module-attribute 

EnvironmentName = 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})

CompletionStatus

Bases: NamedTuple

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

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

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

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

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

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

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

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

complete instance-attribute 

complete: bool

success instance-attribute 

success: bool

scored_by class-attribute instance-attribute 

scored_by: Optional[str] = None

Environment

Bases: Protocol

A kind of workcell and the tasks it offers.

Source code in core/src/armnet_core/environment.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@runtime_checkable
class Environment(Protocol):
    """A kind of workcell and the tasks it offers."""

    name: str

    def tasks(self) -> Mapping[str, TaskSpec]:
        """Every task this environment can run, keyed by slug."""

    def connect(
        self,
        config: Mapping[str, Any],
        *,
        task: Optional[str],
        report_progress: Optional[Callable[[str], None]] = None,
    ) -> Optional[InstrumentedCell]:
        """Open a session, or return ``None`` if this cell has no hardware.

        ``config`` is the cell's environment block, passed through untouched
        from its configuration file.
        """

name instance-attribute 

name: str

tasks 

tasks() -> Mapping[str, TaskSpec]

Every task this environment can run, keyed by slug.

Source code in core/src/armnet_core/environment.py 
173
174
def tasks(self) -> Mapping[str, TaskSpec]:
    """Every task this environment can run, keyed by slug."""

connect 

connect(config: Mapping[str, Any], *, task: Optional[str], report_progress: Optional[Callable[[str], None]] = None) -> Optional[InstrumentedCell]

Open a session, or return None if this cell has no hardware.

config is the cell's environment block, passed through untouched from its configuration file.

Source code in core/src/armnet_core/environment.py 
176
177
178
179
180
181
182
183
184
185
186
187
def connect(
    self,
    config: Mapping[str, Any],
    *,
    task: Optional[str],
    report_progress: Optional[Callable[[str], None]] = None,
) -> Optional[InstrumentedCell]:
    """Open a session, or return ``None`` if this cell has no hardware.

    ``config`` is the cell's environment block, passed through untouched
    from its configuration file.
    """

EnvironmentNotFound

Bases: LookupError

Raised when no installed package provides the named environment.

Source code in core/src/armnet_core/environment.py 
190
191
class EnvironmentNotFound(LookupError):
    """Raised when no installed package provides the named environment."""

InstrumentedCell

Bases: Protocol

A live session against one cell's instrumentation, for one job.

Implementations own everything about their kind of workcell: the transport, how a task's goal is expressed, and how the scene is restored between episodes. Job code sees only this interface, through ctx.cell.

No method here may raise because the hardware misbehaved. Instrumentation is an aid to a rollout, never the thing that fails one, so a silent panel reports "nothing known" and lets the operator decide.

Source code in core/src/armnet_core/environment.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@runtime_checkable
class InstrumentedCell(Protocol):
    """A live session against one cell's instrumentation, for one job.

    Implementations own everything about their kind of workcell: the transport,
    how a task's goal is expressed, and how the scene is restored between
    episodes. Job code sees only this interface, through ``ctx.cell``.

    No method here may raise because the hardware misbehaved. Instrumentation
    is an aid to a rollout, never the thing that fails one, so a silent panel
    reports "nothing known" and lets the operator decide.
    """

    def instrument(self, robot: Any) -> Any:
        """Return ``robot``, wrapped if this environment records alongside it.

        The wrapper must not change the observation the policy sees. Readings
        belong in a sidecar, not in ``observation.state``, or a policy trained
        without the instrumentation will behave differently with it attached.
        """

    def readings(self) -> Mapping[str, Reading]:
        """Current value of every channel, empty if nothing has been heard."""

    def begin_episode(self) -> Iterable[str]:
        """Open an episode window and report anything wrong with the scene.

        Returns human-readable problems — a goal already satisfied before the
        rollout starts, say — which make the episode unscorable rather than
        failed. An empty iterable means the scene is ready.
        """

    def episode_status(self, *, final: bool = False) -> CompletionStatus:
        """Whether the instrumentation can call this episode yet.

        ``complete=False`` means "no verdict", whether because the goal is
        unmet or because nothing readable bears on it. Callers fall back to
        another judge; they cannot distinguish, and should not need to.

        ``final=True`` says the episode is out of time and asks for a last
        word. An unmet goal is then a failure rather than an abstention: an
        operator should not be asked to confirm a failure the instrumentation
        can see. Channels that read nothing still abstain.
        """

    def reset_scene(self, *, confirm: bool, reset_cell: Callable[[bool], None]) -> None:
        """Restore the state this task starts from.

        ``reset_cell`` returns the arm to rest and, when passed ``True``, waits
        for an operator. Implementations decide whether a human is needed;
        ``confirm=True`` means the caller has already decided one is.
        """

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

    def record_frame(self) -> None:
        """Buffer the current readings against the frame just recorded."""

    def commit_episode(self, episode_index: int) -> None:
        """Persist buffered readings for a saved episode."""

    def discard_episode(self) -> None:
        """Drop buffered readings for an episode that will not be saved."""

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

instrument 

instrument(robot: Any) -> Any

Return robot, wrapped if this environment records alongside it.

The wrapper must not change the observation the policy sees. Readings belong in a sidecar, not in observation.state, or a policy trained without the instrumentation will behave differently with it attached.

Source code in core/src/armnet_core/environment.py 
111
112
113
114
115
116
117
def instrument(self, robot: Any) -> Any:
    """Return ``robot``, wrapped if this environment records alongside it.

    The wrapper must not change the observation the policy sees. Readings
    belong in a sidecar, not in ``observation.state``, or a policy trained
    without the instrumentation will behave differently with it attached.
    """

readings 

readings() -> Mapping[str, Reading]

Current value of every channel, empty if nothing has been heard.

Source code in core/src/armnet_core/environment.py 
119
120
def readings(self) -> Mapping[str, Reading]:
    """Current value of every channel, empty if nothing has been heard."""

begin_episode 

begin_episode() -> Iterable[str]

Open an episode window and report anything wrong with the scene.

Returns human-readable problems — a goal already satisfied before the rollout starts, say — which make the episode unscorable rather than failed. An empty iterable means the scene is ready.

Source code in core/src/armnet_core/environment.py 
122
123
124
125
126
127
128
def begin_episode(self) -> Iterable[str]:
    """Open an episode window and report anything wrong with the scene.

    Returns human-readable problems — a goal already satisfied before the
    rollout starts, say — which make the episode unscorable rather than
    failed. An empty iterable means the scene is ready.
    """

episode_status 

episode_status(*, final: bool = False) -> CompletionStatus

Whether the instrumentation can call this episode yet.

complete=False means "no verdict", whether because the goal is unmet or because nothing readable bears on it. Callers fall back to another judge; they cannot distinguish, and should not need to.

final=True says the episode is out of time and asks for a last word. An unmet goal is then a failure rather than an abstention: an operator should not be asked to confirm a failure the instrumentation can see. Channels that read nothing still abstain.

Source code in core/src/armnet_core/environment.py 
130
131
132
133
134
135
136
137
138
139
140
141
def episode_status(self, *, final: bool = False) -> CompletionStatus:
    """Whether the instrumentation can call this episode yet.

    ``complete=False`` means "no verdict", whether because the goal is
    unmet or because nothing readable bears on it. Callers fall back to
    another judge; they cannot distinguish, and should not need to.

    ``final=True`` says the episode is out of time and asks for a last
    word. An unmet goal is then a failure rather than an abstention: an
    operator should not be asked to confirm a failure the instrumentation
    can see. Channels that read nothing still abstain.
    """

reset_scene 

reset_scene(*, confirm: bool, reset_cell: Callable[[bool], None]) -> None

Restore the state this task starts from.

reset_cell returns the arm to rest and, when passed True, waits for an operator. Implementations decide whether a human is needed; confirm=True means the caller has already decided one is.

Source code in core/src/armnet_core/environment.py 
143
144
145
146
147
148
149
def reset_scene(self, *, confirm: bool, reset_cell: Callable[[bool], None]) -> None:
    """Restore the state this task starts from.

    ``reset_cell`` returns the arm to rest and, when passed ``True``, waits
    for an operator. Implementations decide whether a human is needed;
    ``confirm=True`` means the caller has already decided one is.
    """

attach_dataset 

attach_dataset(dataset_root: Any) -> None

Begin recording readings beside a dataset being written.

Source code in core/src/armnet_core/environment.py 
151
152
def attach_dataset(self, dataset_root: Any) -> None:
    """Begin recording readings beside a dataset being written."""

record_frame 

record_frame() -> None

Buffer the current readings against the frame just recorded.

Source code in core/src/armnet_core/environment.py 
154
155
def record_frame(self) -> None:
    """Buffer the current readings against the frame just recorded."""

commit_episode 

commit_episode(episode_index: int) -> None

Persist buffered readings for a saved episode.

Source code in core/src/armnet_core/environment.py 
157
158
def commit_episode(self, episode_index: int) -> None:
    """Persist buffered readings for a saved episode."""

discard_episode 

discard_episode() -> None

Drop buffered readings for an episode that will not be saved.

Source code in core/src/armnet_core/environment.py 
160
161
def discard_episode(self) -> None:
    """Drop buffered readings for an episode that will not be saved."""

close 

close() -> None

Release the connection.

Source code in core/src/armnet_core/environment.py 
163
164
def close(self) -> None:
    """Release the connection."""

Reading dataclass

One instrumentation channel's current value.

continuous separates channels that sweep through values as they move, such as a slider or a dial, from ones that jump between a few states, such as a switch. Anything watching the instrumentation has to treat those differently: a switch is worth reporting on every change, while a dial being turned would otherwise report continuously.

Source code in core/src/armnet_core/environment.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass(frozen=True)
class Reading:
    """One instrumentation channel's current value.

    ``continuous`` separates channels that sweep through values as they move,
    such as a slider or a dial, from ones that jump between a few states, such
    as a switch. Anything watching the instrumentation has to treat those
    differently: a switch is worth reporting on every change, while a dial
    being turned would otherwise report continuously.
    """

    name: str
    kind: str
    value: float
    continuous: bool = False
    label: Optional[str] = None

    def describe(self) -> str:
        return f"{self.name}: {self.label if self.label is not None else self.value}"

name instance-attribute 

name: str

kind instance-attribute 

kind: str

value instance-attribute 

value: float

continuous class-attribute instance-attribute 

continuous: bool = False

label class-attribute instance-attribute 

label: Optional[str] = None

describe 

describe() -> str
Source code in core/src/armnet_core/environment.py 
82
83
def describe(self) -> str:
    return f"{self.name}: {self.label if self.label is not None else self.value}"

TaskSpec dataclass

A task an environment offers.

instruction is the natural-language form handed to a policy, and is what the platform stores as the task description.

Source code in core/src/armnet_core/environment.py 
86
87
88
89
90
91
92
93
94
95
@dataclass(frozen=True)
class TaskSpec:
    """A task an environment offers.

    ``instruction`` is the natural-language form handed to a policy, and is
    what the platform stores as the task description.
    """

    slug: str
    instruction: str

slug instance-attribute 

slug: str

instruction instance-attribute 

instruction: str

CellRuntimeStatus

Bases: str, Enum

Current availability state for a robot cell.

Source code in core/src/armnet_core/models.py 
61
62
63
64
65
66
67
class CellRuntimeStatus(str, Enum):
    """Current availability state for a robot cell."""

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

AVAILABLE class-attribute instance-attribute 

AVAILABLE = 'available'

OCCUPIED class-attribute instance-attribute 

OCCUPIED = 'occupied'

MAINTENANCE class-attribute instance-attribute 

MAINTENANCE = 'maintenance'

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 
369
370
371
372
373
374
375
376
377
378
379
class CellStatus(BaseModel):
    """Current status for one cell as reported by recent heartbeats."""

    cell_id: str
    embodiment: Embodiment
    # What the cell is set up to work on. It can run any task belonging to this,
    # so the cell reports the environment and the job carries the task.
    environment: Environment
    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

environment instance-attribute 

environment: Environment

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.

Callers ask by task, which is what they think in; the orchestrator resolves it to the environment that can run it and matches cells on that. Both are echoed back. task is None when the summary aggregates across the whole embodiment, which is what a task-less job needs.

Source code in core/src/armnet_core/models.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
class CellStatusSummary(BaseModel):
    """Aggregate availability for an embodiment, optionally scoped to a task.

    Callers ask by ``task``, which is what they think in; the orchestrator
    resolves it to the environment that can run it and matches cells on that.
    Both are echoed back. ``task`` is ``None`` when the summary aggregates
    across the whole embodiment, which is what a task-less job needs.
    """

    embodiment: Embodiment
    task: Optional[Task] = None
    environment: Optional[Environment] = 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

environment class-attribute instance-attribute 

environment: Optional[Environment] = 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 
640
641
642
643
644
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 
647
648
649
650
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 
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
class Job(BaseModel):
    """The orchestrator's view of a job (response body of GET /jobs/{id})."""

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

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

id class-attribute instance-attribute 

id: str = Field(default_factory=_new_job_id)

spec instance-attribute 

spec: JobSpec

status class-attribute instance-attribute 

status: JobStatus = JobStatus.SUBMITTED

created_at class-attribute instance-attribute 

created_at: datetime = Field(default_factory=_utcnow)

updated_at class-attribute instance-attribute 

updated_at: datetime = Field(default_factory=_utcnow)

cell_id class-attribute instance-attribute 

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

result class-attribute instance-attribute 

result: Optional[JobResult] = None

dispatched class-attribute instance-attribute 

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

is_terminal 

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

JobResult

Bases: BaseModel

Terminal result published by a cell on the results subject.

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

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

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

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

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

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

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

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

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

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

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

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

status class-attribute instance-attribute 

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

exit_code class-attribute instance-attribute 

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

stdout class-attribute instance-attribute 

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

stderr class-attribute instance-attribute 

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

error class-attribute instance-attribute 

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

traceback class-attribute instance-attribute 

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

return_value class-attribute instance-attribute 

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

started_at class-attribute instance-attribute 

started_at: Optional[datetime] = None

finished_at class-attribute instance-attribute 

finished_at: Optional[datetime] = None

raise_for_status 

raise_for_status() -> None

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

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

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

For successful results this is a no-op.

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

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

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

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

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

JobSpec

Bases: BaseModel

The fields a client supplies when creating a job.

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

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

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

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

image class-attribute instance-attribute 

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

args class-attribute instance-attribute 

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

embodiment class-attribute instance-attribute 

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

task class-attribute instance-attribute 

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

timeout_seconds class-attribute instance-attribute 

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

secrets class-attribute instance-attribute 

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

detach class-attribute instance-attribute 

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

username class-attribute instance-attribute 

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

JobStatus

Bases: str, Enum

Job lifecycle states.

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

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

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

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

SUBMITTED class-attribute instance-attribute 

SUBMITTED = 'submitted'

QUEUED class-attribute instance-attribute 

QUEUED = 'queued'

DISPATCHED class-attribute instance-attribute 

DISPATCHED = 'dispatched'

RUNNING class-attribute instance-attribute 

RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute 

SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute 

FAILED = 'failed'

TIMEOUT class-attribute instance-attribute 

TIMEOUT = 'timeout'

CANCELLED class-attribute instance-attribute 

CANCELLED = 'cancelled'

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 
533
534
535
536
537
538
539
540
541
542
543
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 
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
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 
546
547
548
549
550
551
552
553
554
555
556
557
558
class RemoteExecutionError(RuntimeError):
    """Raised by :meth:`JobResult.raise_for_status` for non-succeeded results.

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

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

result instance-attribute 

result = result

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 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
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
    # The task being run right now, stamped in per job. None outside a job: the
    # cell is set up for an environment, not a task, and hosts all of them.
    task: Optional[Task] = None
    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
    current_gate_enabled: bool = True
    # Data interface params
    tcp_read_timeout: float = 10.0
    jpeg_quality: int = 0
    # Cell infra
    shelly_timeout: float = 2.0
    # What this cell is set up to work on, and that environment's own settings.
    # The settings stay an opaque dict: only the package implementing the
    # environment understands them, and armnet-core must remain pydantic-only.
    # They ride inline rather than as a path because RobotCellConfig is
    # serialized whole into ARMNET_CELL_CONFIG, so an inline mapping reaches the
    # container with no bind mount.
    environment: Optional[str] = None
    environment_config: dict[str, Any] = Field(default_factory=dict)
    # Workcell lightbox: HTTP-controlled LED dimmer on the cell's own AP (e.g.
    # "http://lightbox.local"). Only the edge host can reach it, so the cell
    # routes brightness through the edge connector at job start. None = no
    # lightbox on this cell. Brightness/freq/fade are cell defaults (JSON);
    # jobs may override brightness via args.lightbox_brightness.
    lightbox_url: Optional[str] = None
    lightbox_default_brightness: int = 30  # percent 0-100
    lightbox_default_frequency: int = 2000  # PWM Hz
    lightbox_default_fade: int = 1000  # ms ramp on /set

    @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)
            flat["current_gate_enabled"] = robot.get("current_gate_enabled", True)

        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)

        # The environment's settings live in a block named after it and are
        # passed through whole. Listing the keys we expect, as every other block
        # here does, is exactly how a new setting gets silently dropped on its
        # way to the cell.
        environment = data.pop("environment", None)
        if environment:
            flat["environment"] = environment
            flat["environment_config"] = data.pop(environment, None) or {}

        lightbox = data.pop("lightbox", None)
        if isinstance(lightbox, dict):
            flat["lightbox_url"] = lightbox.get("url")
            if lightbox.get("default_brightness") is not None:
                flat["lightbox_default_brightness"] = lightbox["default_brightness"]
            if lightbox.get("default_frequency") is not None:
                flat["lightbox_default_frequency"] = lightbox["default_frequency"]
            if lightbox.get("default_fade") is not None:
                flat["lightbox_default_fade"] = lightbox["default_fade"]

        policy = data.pop("policy", None)
        if isinstance(policy, dict):
            # NOTE: ``task`` / ``language_instruction`` are intentionally NOT read
            # from the config file. They belong to the job, not the cell: a cell
            # hosts every task its environment offers and is told which one this
            # is when the job arrives. Any ``policy.task`` /
            # ``policy.language_instruction`` left in a JSON file is ignored so a
            # stale copy can't override it.
            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", "environment", "lightbox")):
            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: Optional[Task] = None

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

current_gate_enabled class-attribute instance-attribute 

current_gate_enabled: bool = True

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

environment class-attribute instance-attribute 

environment: Optional[str] = None

environment_config class-attribute instance-attribute 

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

lightbox_url class-attribute instance-attribute 

lightbox_url: Optional[str] = None

lightbox_default_brightness class-attribute instance-attribute 

lightbox_default_brightness: int = 30

lightbox_default_frequency class-attribute instance-attribute 

lightbox_default_frequency: int = 2000

lightbox_default_fade class-attribute instance-attribute 

lightbox_default_fade: int = 1000

from_file classmethod 

from_file(path: str | Path) -> 'RobotCellConfig'
Source code in core/src/armnet_core/models.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
@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 
678
679
680
681
682
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 
666
667
668
669
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 
672
673
674
675
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 
653
654
655
656
657
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 
660
661
662
663
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 
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
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 
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
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 
625
626
627
628
629
630
631
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 
634
635
636
637
class WhoAmI(BaseModel):
    """Authenticated API identity."""

    username: str

username instance-attribute 

username: str

available_environments 

available_environments() -> list[str]

Names of every environment provided by an installed package.

Source code in core/src/armnet_core/environment.py 
200
201
202
203
def available_environments() -> list[str]:
    """Names of every environment provided by an installed package."""

    return sorted({entry.name for entry in _entry_points()})

environment_for 

environment_for(name: str) -> Environment

Load the environment registered under name.

Loading is deferred to here so that importing this module costs nothing: an environment's drivers are only imported by a process that actually intends to talk to one.

Source code in core/src/armnet_core/environment.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def environment_for(name: str) -> Environment:
    """Load the environment registered under ``name``.

    Loading is deferred to here so that importing this module costs nothing:
    an environment's drivers are only imported by a process that actually
    intends to talk to one.
    """

    for entry in _entry_points():
        if entry.name == name:
            return entry.load()()
    installed = available_environments()
    raise EnvironmentNotFound(
        f"no environment named {name!r} is installed"
        + (f" (installed: {', '.join(installed)})" if installed else "")
    )

task_environment 

task_environment(task: Optional[str]) -> Optional[str]

Name the environment that owns task, or None if unowned.

Used where the mapping is needed without a database — chiefly tests and local runs. The orchestrator resolves through the tasks table instead, so routing never depends on which environment packages it has installed.

Source code in core/src/armnet_core/environment.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def task_environment(task: Optional[str]) -> Optional[str]:
    """Name the environment that owns ``task``, or ``None`` if unowned.

    Used where the mapping is needed without a database — chiefly tests and
    local runs. The orchestrator resolves through the ``tasks`` table instead,
    so routing never depends on which environment packages it has installed.
    """

    if not task:
        return None
    for name in available_environments():
        if task in environment_for(name).tasks():
            return name
    return None

armnet_core.enums

Embodiment, environment and task wire-type aliases.

These are stable string slugs that appear in HTTP bodies, NATS subjects, env vars injected into customer containers, and cell configuration (for example "lerobot/so-101", "busybox", "busybox_left_toggle_down").

The three name different things:

Embodiment is the robot — what arms the cell has.

Environment is what the robot is working on: an instrumented BusyBox panel, a table of wooden blocks. A cell is assigned one, and can then run any task that environment offers without being reconfigured.

Task is a single job to perform within an environment. Users request a task; the platform resolves it to an environment to find a cell that can run it.

Valid embodiments and environments live in the orchestrator's Postgres database (see db/), which routing treats as the source of truth so it never depends on which environment packages happen to be installed. Task behaviour — the goal, how it is scored, how the scene is restored — lives in the environment's own package, because it is inseparable from that environment's instrumentation. The database mirrors those tasks for validation and routing.

Embodiment module-attribute 

Embodiment = str

Environment module-attribute 

Environment = str

Task module-attribute 

Task = str

DEFAULT_EMBODIMENT module-attribute 

DEFAULT_EMBODIMENT: Embodiment = 'lerobot/so-101'

DEFAULT_ENVIRONMENT module-attribute 

DEFAULT_ENVIRONMENT: Environment = 'busybox'

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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class JobStatus(str, Enum):
    """Job lifecycle states.

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

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

SUBMITTED class-attribute instance-attribute 

SUBMITTED = 'submitted'

QUEUED class-attribute instance-attribute 

QUEUED = 'queued'

DISPATCHED class-attribute instance-attribute 

DISPATCHED = 'dispatched'

RUNNING class-attribute instance-attribute 

RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute 

SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute 

FAILED = 'failed'

TIMEOUT class-attribute instance-attribute 

TIMEOUT = 'timeout'

CANCELLED class-attribute instance-attribute 

CANCELLED = 'cancelled'

CellRuntimeStatus

Bases: str, Enum

Current availability state for a robot cell.

Source code in core/src/armnet_core/models.py 
61
62
63
64
65
66
67
class CellRuntimeStatus(str, Enum):
    """Current availability state for a robot cell."""

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

AVAILABLE class-attribute instance-attribute 

AVAILABLE = 'available'

OCCUPIED class-attribute instance-attribute 

OCCUPIED = 'occupied'

MAINTENANCE class-attribute instance-attribute 

MAINTENANCE = 'maintenance'

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 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
class JobSpec(BaseModel):
    """The fields a client supplies when creating a job.

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

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

image class-attribute instance-attribute 

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

args class-attribute instance-attribute 

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

embodiment class-attribute instance-attribute 

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

task class-attribute instance-attribute 

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

timeout_seconds class-attribute instance-attribute 

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

secrets class-attribute instance-attribute 

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

detach class-attribute instance-attribute 

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

username class-attribute instance-attribute 

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

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 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
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
    # The task being run right now, stamped in per job. None outside a job: the
    # cell is set up for an environment, not a task, and hosts all of them.
    task: Optional[Task] = None
    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
    current_gate_enabled: bool = True
    # Data interface params
    tcp_read_timeout: float = 10.0
    jpeg_quality: int = 0
    # Cell infra
    shelly_timeout: float = 2.0
    # What this cell is set up to work on, and that environment's own settings.
    # The settings stay an opaque dict: only the package implementing the
    # environment understands them, and armnet-core must remain pydantic-only.
    # They ride inline rather than as a path because RobotCellConfig is
    # serialized whole into ARMNET_CELL_CONFIG, so an inline mapping reaches the
    # container with no bind mount.
    environment: Optional[str] = None
    environment_config: dict[str, Any] = Field(default_factory=dict)
    # Workcell lightbox: HTTP-controlled LED dimmer on the cell's own AP (e.g.
    # "http://lightbox.local"). Only the edge host can reach it, so the cell
    # routes brightness through the edge connector at job start. None = no
    # lightbox on this cell. Brightness/freq/fade are cell defaults (JSON);
    # jobs may override brightness via args.lightbox_brightness.
    lightbox_url: Optional[str] = None
    lightbox_default_brightness: int = 30  # percent 0-100
    lightbox_default_frequency: int = 2000  # PWM Hz
    lightbox_default_fade: int = 1000  # ms ramp on /set

    @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)
            flat["current_gate_enabled"] = robot.get("current_gate_enabled", True)

        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)

        # The environment's settings live in a block named after it and are
        # passed through whole. Listing the keys we expect, as every other block
        # here does, is exactly how a new setting gets silently dropped on its
        # way to the cell.
        environment = data.pop("environment", None)
        if environment:
            flat["environment"] = environment
            flat["environment_config"] = data.pop(environment, None) or {}

        lightbox = data.pop("lightbox", None)
        if isinstance(lightbox, dict):
            flat["lightbox_url"] = lightbox.get("url")
            if lightbox.get("default_brightness") is not None:
                flat["lightbox_default_brightness"] = lightbox["default_brightness"]
            if lightbox.get("default_frequency") is not None:
                flat["lightbox_default_frequency"] = lightbox["default_frequency"]
            if lightbox.get("default_fade") is not None:
                flat["lightbox_default_fade"] = lightbox["default_fade"]

        policy = data.pop("policy", None)
        if isinstance(policy, dict):
            # NOTE: ``task`` / ``language_instruction`` are intentionally NOT read
            # from the config file. They belong to the job, not the cell: a cell
            # hosts every task its environment offers and is told which one this
            # is when the job arrives. Any ``policy.task`` /
            # ``policy.language_instruction`` left in a JSON file is ignored so a
            # stale copy can't override it.
            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", "environment", "lightbox")):
            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: Optional[Task] = None

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

current_gate_enabled class-attribute instance-attribute 

current_gate_enabled: bool = True

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

environment class-attribute instance-attribute 

environment: Optional[str] = None

environment_config class-attribute instance-attribute 

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

lightbox_url class-attribute instance-attribute 

lightbox_url: Optional[str] = None

lightbox_default_brightness class-attribute instance-attribute 

lightbox_default_brightness: int = 30

lightbox_default_frequency class-attribute instance-attribute 

lightbox_default_frequency: int = 2000

lightbox_default_fade class-attribute instance-attribute 

lightbox_default_fade: int = 1000

from_file classmethod 

from_file(path: str | Path) -> 'RobotCellConfig'
Source code in core/src/armnet_core/models.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
@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 
369
370
371
372
373
374
375
376
377
378
379
class CellStatus(BaseModel):
    """Current status for one cell as reported by recent heartbeats."""

    cell_id: str
    embodiment: Embodiment
    # What the cell is set up to work on. It can run any task belonging to this,
    # so the cell reports the environment and the job carries the task.
    environment: Environment
    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

environment instance-attribute 

environment: Environment

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.

Callers ask by task, which is what they think in; the orchestrator resolves it to the environment that can run it and matches cells on that. Both are echoed back. task is None when the summary aggregates across the whole embodiment, which is what a task-less job needs.

Source code in core/src/armnet_core/models.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
class CellStatusSummary(BaseModel):
    """Aggregate availability for an embodiment, optionally scoped to a task.

    Callers ask by ``task``, which is what they think in; the orchestrator
    resolves it to the environment that can run it and matches cells on that.
    Both are echoed back. ``task`` is ``None`` when the summary aggregates
    across the whole embodiment, which is what a task-less job needs.
    """

    embodiment: Embodiment
    task: Optional[Task] = None
    environment: Optional[Environment] = 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

environment class-attribute instance-attribute 

environment: Optional[Environment] = 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 
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
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 
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
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 
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
class JobResult(BaseModel):
    """Terminal result published by a cell on the results subject.

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

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

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

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

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

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

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

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

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

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

status class-attribute instance-attribute 

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

exit_code class-attribute instance-attribute 

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

stdout class-attribute instance-attribute 

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

stderr class-attribute instance-attribute 

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

error class-attribute instance-attribute 

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

traceback class-attribute instance-attribute 

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

return_value class-attribute instance-attribute 

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

started_at class-attribute instance-attribute 

started_at: Optional[datetime] = None

finished_at class-attribute instance-attribute 

finished_at: Optional[datetime] = None

raise_for_status 

raise_for_status() -> None

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

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

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

For successful results this is a no-op.

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

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

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

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

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

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 
533
534
535
536
537
538
539
540
541
542
543
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 
546
547
548
549
550
551
552
553
554
555
556
557
558
class RemoteExecutionError(RuntimeError):
    """Raised by :meth:`JobResult.raise_for_status` for non-succeeded results.

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

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

result instance-attribute 

result = result

Job

Bases: BaseModel

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

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

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

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

id class-attribute instance-attribute 

id: str = Field(default_factory=_new_job_id)

spec instance-attribute 

spec: JobSpec

status class-attribute instance-attribute 

status: JobStatus = JobStatus.SUBMITTED

created_at class-attribute instance-attribute 

created_at: datetime = Field(default_factory=_utcnow)

updated_at class-attribute instance-attribute 

updated_at: datetime = Field(default_factory=_utcnow)

cell_id class-attribute instance-attribute 

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

result class-attribute instance-attribute 

result: Optional[JobResult] = None

dispatched class-attribute instance-attribute 

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

is_terminal 

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

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 
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
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 
625
626
627
628
629
630
631
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 
634
635
636
637
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 
640
641
642
643
644
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 
647
648
649
650
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 
653
654
655
656
657
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 
660
661
662
663
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 
666
667
668
669
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 
672
673
674
675
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 
678
679
680
681
682
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'