# Device, stream and policy adapters

These adapters support an agent-requested experiment API. Hardware is a resource
behind the experiment's capability/state/compute contract; it must still be
individually admitted. The [planned experiment journey](ROOM_API.md#planned-experiment-journey-and-authorization)
composes existing booking/job authority. It does not change the SDK interfaces
documented here or grant model code spending or physical control authority.

SDK 0.1.7 adds hardware-neutral **local Python interfaces**, with concrete
SO100 and paired NexArm mappings. Hardware-specific names, limits and wire
versions live in adapters. The shared types have no arm count, joint order,
mandatory robot state, fixed camera or model runtime. This follows
[ADR 0057](architecture/0057-room-capabilities-and-independent-streaming-sdk.md).
SDK 0.1.8 adds a [generic execution loop](SDK_EXECUTION.md) over these interfaces,
with separate stream readers and an injected control backend. The first backend
composes the existing paused booked-room client and its original lifecycle.
The [placement plan](SDK_EXECUTION.md#planned-execution-placement) keeps customer
decisions separate from local drivers/controllers whether the customer runtime
is remote or, in the future rental mode, on an onsite workstation. Streams,
events and final results remain independent of motion authority.

These interfaces do not introduce a server execution protocol or expand an old
one. `actuator.robot-client@1`, its NexArm forecast/feedback runner and all old
records retain their meanings. Room admission, booking, retry, applied-command
evidence, publication and credit authority remain the existing services.

## What is implemented

| Layer | Shared interface | Current implementation |
| --- | --- | --- |
| Discovery/access | `RoomDirectory`, `ObservationClient` | Existing project-scoped published descriptions/docs and original RGB reads. Public `@1` descriptions still describe the currently admitted simulation rooms. |
| Controls | `ControlDescription`, `Channel`, `NamedTargets`, `DeviceState`, `DeviceAdapter`, `AdapterRegistry` | Explicit local target/decision mapping for SO100 capture `@4`, SO100 session `@5` and paused NexArm `@1`. |
| Observations | `StreamDescription`, `StreamReader`, `StreamSample`, `StreamRead`, `StreamMux` | `RGBStream` adapts existing original PNGs from all five current capabilities. Sensor/inference readers have fixture coverage only. |
| Customer models | `Policy.prepare`, `Policy.predict`, `ExecutionLoop` | Prepare before booking; finite prediction, source freshness and actual application through an explicit control backend. Customer code supplies preprocessing, model, instruction and decoder. |
| Execution | `ControlSession`, `PausedRoomControl` | Injected existing starter retains pinned authority, durable retry, application, Close and verified export for the three paused capabilities. Streams remain independent. |
| Paused NexArm sensing (0.1.9) | `NexArmSensingStream`, `sample_for_state` | Existing original three-camera/joint packet through an injected read-only call; exact decision/drive association. No image/model dependencies or new server schema. |
| Legacy continuous control | Existing `RobotClient`, `RunConfig`, `PolicyOutput`, `LeRobotPolicy` | Unchanged NexArm-specific runner with its existing timing, forecasting and reconnect behavior. |

SDK 0.1.10 keeps the generic stream values in `stream_contracts.py`, collection
in `streams.py` and original-PNG adaptation in `rgb_stream.py`. Shared identifiers
and generic validation live in `primitives.py`; device controls stay in
`devices.py`. Prior imports continue to work. See the
[module map](../src/physical_ai_lab/workshop/robotics/sdk/README.md) when adding
an adapter; source layout changes introduce no new wire or hardware admission.

Core SDK installation needs HTTP/schema dependencies only. Importing shared
interfaces or paused adapters imports no server, native engine, image decoder,
NumPy, Torch, ROS or legacy `RobotClient`. Media/ML/device stacks belong in the
individual integrations that need them.

## Named controls

`ControlDescription` identifies a control mode, exact command schema, clock,
period and row limit. Every channel declares its name, quantity, unit, optional
frame and finite ordered limits. There is no global six-/twelve-channel vector.
Mixed angle/distance channels and other widths are exercised as contract
fixtures; that is not admission of a new robot or control mode.

`NamedTargets` contains absolute rows with explicit channel names, units, frames
and period. `control.rows(targets)` validates them and orders values exactly as
declared. It rejects missing/extra channels, nonfinite/boolean values, unknown
fields, out-of-range targets and changed units/frames/rates. It does not infer a
joint mapping, convert degrees, resample, clip, integrate model deltas or assume
that gripper angles mean jaw aperture. Those decisions belong to a qualified
device adapter and the customer's explicit decoder.

The current paused adapters expose one `joint_positions` mode in radians. Use
the actual booked interface and full original native decision:

```python
from actuator_sdk import NamedTargets
from actuator_sdk.joint_adapters import paused_position_adapters

registry = paused_position_adapters()
adapter = registry.resolve(interface)
control = adapter.describe(interface)
state = adapter.state(interface, live_decision)

# Your decoder supplies absolute named targets in these units/frames.
# measured positions are not an implicit anchor for model output deltas.
output = NamedTargets(
    control_id=control.id,
    period_s=control.period_s,
    units={c.name: c.unit for c in control.channels},
    frames={c.name: c.frame for c in control.channels},
    values=decoded_named_rows,
)
command = adapter.command(
    interface, live_decision, output,
    accept_by_unix=finite_original_server_deadline,
    idempotency_key=original_retry_key,
)
```

The helper is pure: it sends no command and grants no authority. Submit through
the existing booked-room starter's durable `mutate('command', command)` path or
an equivalent caller-owned exact-retry journal over the existing command API.
Persist the original request before sending; recover uncertain outcomes with
that exact request. Retain its deadline and key on retries. Never interpret
acceptance as native application or silently construct a replacement after an
unknown transport outcome. `PausedRoomControl` composes these existing starter
responsibilities; `ExecutionLoop` does not introduce another job lifecycle.

The adapter verifies the admitted capability/schema, original observation
digest, project/job agreement, names/limits/clock and a finite server deadline.
Unsupported execution-affecting control fields fail. Existing server checks
still own current scope, lease, calendar, sequence, slew and action budgets.
`DeviceState.measured` preserves actual finite joint readings. Current paused
decisions omit drive targets, so `drive_targets` is **None**. A delta decoder
must obtain a separately verified, source-associated drive anchor; substituting
measured positions would change its semantics.

## Customer deltas and a verified drive anchor

SDK 0.1.9 supplies one concrete reader for the already-admitted paused NexArm
packet. It preserves the full original sample, cameras, units, timestamps and
digests; it grants no control authority and imports no server, engine, image
decoder or model runtime. The caller owns its read-only transport:

```python
from actuator_sdk.nexarm_sensing import NexArmSensingStream, sample_for_state

# read_sensing is an async GET of the existing project's
# /live/{job_id}/interface/sensing route, using a scoped read credential.
stream = NexArmSensingStream(project_id, job_id, read_sensing)

# In the customer's policy, after ExecutionLoop checks binding and freshness:
sample = sample_for_state(streams['nexarm_sensing'], state)
joint = sample['joint_state']
drive = dict(zip(joint['names'], joint['position_targets_rad']))
```

Use `NexArmSensingStream.describe()` during pre-booking preparation and require
fresh samples matching the control step. The helper additionally checks the
sample's decision digest and measured coordinates against that exact state,
and checks that wrapper/source metadata matches the unchanged payload. Missing,
altered, foreign or stale data must not become a fresh prediction. Project/job
and generation fencing remain ExecutionLoop/control-backend responsibilities.
`DeviceState.drive_targets` remains unknown in the original paused adapter;
this explicitly separate sensing source supplies the anchor.

The [customer fixture adapter](../tests/fixtures/customer_vla/adapter.py) maps
three native views to its own camera roles, verifies decoded RGB, runs its own
letterbox/normalization and NumPy model, then integrates the model's interleaved
deltas from those drive targets. It rejects invalid paths; it does not clip,
resample, substitute views or interpret another embodiment. Model/image imports
and warmup happen before booking. This is an example for one customer contract,
not a universal VLA decoder.

The [native exercise](results/CUSTOM_VLA_EXECUTION-2026-10-22.md) applies actual
fixture-model outputs over two sequential bookings. The weights are untrained.
Observed source cadence failed its strict 2 Hz minimum, so full experiment
compatibility and learned task performance remain unqualified. A nominal source
target or successful control path cannot establish a delivered minimum rate.

## Independent streams

```python
from actuator_sdk import ObservationClient, RGBStream, StreamMux

async with ObservationClient(url, project_id, read_token) as camera_client:
    mux = StreamMux(project_id, job_id, [
        RGBStream(camera_client, job_id, camera_id)
        for camera_id in declared_camera_ids
    ])
    async for batch in mux.snapshots(timeout_s=10., max_snapshots=20):
        for stream_id, item in batch.items():
            sample = item['read'].sample
            if sample is not None and item['changed']:
                consume_original_frame(stream_id, sample.payload)
                print(sample.source, sample.captured_at,
                      item['skipped_source_sequences'])
```

`RGBStream` preserves the complete original frame payload, PNG bytes/digests,
source schema/sequence/digest, capture/simulation times and calibration metadata.
It uses the existing bounded `actuator.vision-read@1` route and read-only access.
No motion, decoder or mandatory joint state is introduced. Calibration retains
its original limits; the stream does not fabricate optical/world transforms.

`StreamMux` accepts 1–32 uniquely named **explicitly supplied** readers. It reads
them concurrently under a finite timeout including lock wait, cancels pending
reader work on timeout/failure, and keeps one cursor per stream. It rejects
cross-project/job/stream data, changed source schemas, backwards sequences and
changed digests for an existing sequence. Repeated samples are marked unchanged;
gaps are counted separately. A failed batch does not partly advance cursors.
Missing samples and terminal reads stay explicit. Polling bounds do not renew
room authority or close the owned experiment.

This is latest-only collection, **not exposure synchronization**. Camera clocks
and actual cadence remain source-specific. It does not negotiate JPEG/WebRTC,
promise delivered minimum rates, add a camera, or turn disconnection into a
physical stop. After restoring access/connectivity, a caller can deliberately
read again through the same mux; original sequence checks remain. Generation
and motion reconnection rules remain the control adapter's responsibility.

## Adding a sensor or inference reader

Implement `StreamReader.description` and `async latest() -> StreamRead` in an
explicit local module. The adapter validates its exact payload schema, versions,
units, coordinates, size bounds and source integrity. The mux is transport- and
payload-independent; `payload` is deliberately opaque and does not validate or
decode arbitrary sensor data on a reader's behalf. Registering a reader is not
qualification of the producer or authorization to deploy a model.

`StreamDescription` separates measurement, simulation ground truth, model
inference and derived estimates. It names the payload schema, clock domain,
coordinate frame and units. `StreamSample` retains original source identity and
capture time without a compulsory image or joint array.

For model inference, the mux additionally requires producer revision, model and
preprocessing identity, location (`device`, `room_host` or `customer_host`),
nonempty original source references and a finite completion time in the declared
clock domain. Cross-project/job source references fail. Preserve declared
transforms and missing/late/partial outcomes; a score is not assumed to be a
calibrated probability. Fixtures demonstrate depth and device-result association;
no depth camera, detector, IMU or inference server is implemented by this release.

## Adding hardware and policies

Implement `DeviceAdapter` for an exact admitted capability/command schema. Give
the adapter an explicit `ControlDescription`, source-bound `DeviceState` mapping
and command conversion. Register it locally with `AdapterRegistry([adapter])`.
Duplicate registrations and unknown capability/schema combinations fail; the
registry never loads code or installs a driver named by a server response.
Preserve frame/unit conversions and qualify actual rejection, application,
stops, publication and release. A dependency install or matching array width
never admits another device. New physical inventory needs its own platform
adapter, explicit public contract version and native/hardware qualification.

Implement `Policy` structurally; subclassing is optional. `prepare` warms the
customer model against its intended control/stream contract before reserving
capacity. Recheck the actual booked contract before control. `predict` consumes
named stream reads and named state, returning decoded `NamedTargets`, an explicit
`PolicyWait`, or `None` to finish. Keep weights/code on customer compute. There
is no implicit model upload, feature-name inference, action-rate conversion or
hardware transfer.

The [comparative VLA trial](results/VLA_ROOM_SELECTION-2026-10-22.md) remains a
useful integration case: retain every required view, separate measured state
from drive targets, and reject unsupported clocks or replay promises. A future
continuous backend must preserve existing feedback/deadline/unknown-outcome
behavior. The [current paused loop](SDK_EXECUTION.md) preserves those boundaries
through its injected starter; neither interface coverage nor scripted native
execution establishes learned VLA readiness.