The experience data plane

Weight sync moves weights from the trainer to the inference fleet. The data plane moves the other direction’s bulk data — rollout experience and the per-sample tensor columns (log-probs, token IDs, advantages) — between non-colocated fleets, without routing large tensors through Ray’s object store and the driver.

It is optional. With data_plane=None (the default) the in-memory colocated path is used and no external store is involved. When configured, it lets a large (B, S) tensor pass producer→consumer without a driver round-trip.

The boundary

Every call site in algorithms/, experience/, and models/ goes through the DataPlaneClient ABC (data_plane/interfaces.py) — never an adapter directly. That indirection is what makes the implementation swappable. The boundary module imports without tensordict (every TensorDict reference is a deferred annotation), so importing the package stays dependency-light; only the concrete adapters that touch tensor data require it.

Two adapters ship:

impl	Adapter	Use
noop	data_plane/adapters/noop.py	Disabled / passthrough.
transfer_queue	data_plane/adapters/transfer_queue.py	External queue/store with simple or mooncake_cpu backends.

Optional observability middleware (ObservabilityConfig) wraps the chosen adapter to record per-operation metrics.

Control plane vs. data plane

The design separates metadata (cheap, routable) from tensor data (expensive). KVBatchMeta carries per-sample IDs, field names, sequence lengths, and per-sample primitive tags — enough to filter, shard, and balance a batch without fetching any tensor data. For example, the driver can compute a balanced per-DP-rank shard from sequence_lengths alone, and dynamic sampling can drop zero-std rows by inspecting tags, before ever materializing tensors.

KVBatchMeta is a pure value type with subset / slice / concat transforms (each returns a fresh meta), and stamp_tags to mirror scalar columns (std, total_reward, weight_version, …) onto the per-sample sidecar at production time.

The client API

DataPlaneClient methods fall into three intents:

• Task-mediated — for stages that wait on upstream production: register_partition (declare the partition schema and consumer tasks), claim_meta (discover and atomically claim ready samples, advancing a per-task cursor), get_data (resolve a claimed meta to a TensorDict), check_consumption_status.

• Direct-by-key — for stages that already know the uids: put_samples (the producer entrypoint), get_samples (fetch a known slice without advancing any cursor), clear_samples.

• Lifecycle — close (idempotent).

The completion signal is field production, not an explicit “done”: when a stage calls put_samples for a field, the controller flips the production_status bit for that (sample, field) pair, and downstream consumers waiting on that field become unblocked. There is intentionally no mark_consumed.

get_samples and get_data both require an explicit field set — there is no implicit “fetch every field” fallback, because bulk schemas are wide and a silent over-fetch is the most expensive shape the wire can take. Callers must name what they read.