refactor: data movement handled by an independant crate

Cargo.lock

@@ -952,7 +952,10 @@ dependencies = [
 name = "data-plane"
 version = "0.1.0"
 dependencies = [
+ "datastream",
  "libc",
+ "serde",
+ "swactor",
 ]
 
 [[package]]

crates/data-plane/Cargo.toml

@@ -5,6 +5,9 @@ edition = "2024"
 publish = false
 
 [dependencies]
+datastream = { path = "../datastream" }
+serde = { version = "1", features = ["derive"] }
+swactor = { path = "../.." }
 
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"

crates/data-plane/DATA_PLANE_ACTOR_SPEC.md

@@ -0,0 +1,523 @@
+# Data Plane Actor Architecture Specification
+
+**Status:** implemented architecture contract for the `data-plane` crate.
+
+This document describes the behavioral boundary between reusable data-plane actors
+and MVP-specific orchestration/runtime code. It is intentionally architectural: it
+names responsibilities, actor roles, message families, and ownership boundaries
+without prescribing file layout or migration steps.
+
+---
+
+## 1. Purpose
+
+`data-plane` owns the behavior required to move model objects between stages and
+between a node process and its local GPU worker process.
+
+The crate defines the actor protocol for:
+
+- provisioning logical data edges;
+- distinguishing network/wire edges from node-local IPC rings;
+- leasing, installing, readying, faulting, stopping, quiescing, and releasing
+  local rings;
+- binding logical edges to transport endpoints and local worker rings;
+- parsing, validating, sequencing, loading, producing, and reporting objects;
+- gating readiness and object visibility on the correct lifecycle transitions;
+- translating low-level arena, transport, and worker observations into coarse
+  data-plane outcomes.
+
+`mvp-system` uses `data-plane` as a reusable actor subsystem. It provides run
+intent, concrete runtime actor addresses, and MVP-specific report sinks. It does
+not own the fine-grained data-plane state machine.
+
+---
+
+## 2. Core boundary
+
+Swactor actors carry control, lifecycle, and identity messages. Payload bytes do
+not move through actor mailboxes.
+
+Payload bytes move through:
+
+- arena-backed shared-memory rings for node-local IPC;
+- transport byte streams for node-to-node data edges;
+- GPU-worker-owned device allocations for compute-ready objects.
+
+The data-plane actors decide when these byte paths are established, readable,
+writable, faulted, stopped, and safe to release. Runtime adapter actors execute
+concrete effects and report observations back.
+
+---
+
+## 3. Edge and ring model
+
+### 3.1 Wire edge
+
+A wire edge is a logical run-plan connection between a producer endpoint and a
+consumer endpoint.
+
+It carries:
+
+- run identity;
+- edge identity;
+- producer and consumer node identity;
+- edge kind, such as token input, activation, or token output;
+- object contract;
+- transport contract;
+- optional remote endpoint and remote actor identity.
+
+A wire edge answers: "which logical data stream connects these stage endpoints?"
+
+It does not answer: "which local arena offset or worker-process ring is being
+used on this node?"
+
+### 3.2 Local IPC ring
+
+A local IPC ring is a node-local buffer used by a Rust node process and its owned
+GPU worker process.
+
+It carries:
+
+- ring identity allocated by the local arena manager;
+- arena layout and capacity;
+- worker-process generation;
+- local role port, such as input or output;
+- direction relative to the worker process;
+- object contract installed into the worker;
+- quiescence and release state.
+
+A local IPC ring answers: "how does this node exchange bytes with its local
+worker process for a specific edge endpoint?"
+
+It does not answer: "which remote node or distributed route owns the other side
+of the logical edge?"
+
+### 3.3 Binding
+
+A wire edge endpoint may bind to zero or more local resources depending on its
+role:
+
+```text
+inbound wire edge endpoint
+  -> recv transport endpoint
+  -> local worker ingress ring
+  -> device object handle
+
+outbound wire edge endpoint
+  -> local worker egress ring
+  -> send transport endpoint
+
+local-only edge endpoint
+  -> local producer/consumer binding
+  -> optional worker ring
+```
+
+The binding is owned by data-plane state. MVP code may observe the binding only
+through coarse reports such as edge ready, object loaded, object produced, edge
+faulted, and edge stopped.
+
+---
+
+## 4. Actor topology
+
+The target topology is actor-oriented.
+
+### 4.1 Data-plane node actor
+
+One data-plane node actor owns the data-plane state for one local node within one
+active run.
+
+It owns:
+
+- local node identity;
+- run-scoped edge table;
+- mapping from wire edge endpoints to local rings;
+- mapping from ring ids to edge endpoints;
+- object sequence state;
+- device-handle visibility state;
+- data-plane child actor addresses;
+- MVP report sink addresses.
+
+It receives provisioning intent from MVP code and observations from runtime
+adapters. It emits actor messages to arena, worker, transport, and MVP report
+sinks.
+
+### 4.2 Wire edge actor
+
+A wire edge actor owns the lifecycle of one logical edge endpoint on the local
+node.
+
+It owns:
+
+- provisioning state;
+- transport establishment state;
+- send/receive pump readiness;
+- stream faults;
+- logical edge readiness;
+- stop and fault propagation for that edge endpoint.
+
+It does not own worker-process state or arena layout details except through a
+binding supplied by the data-plane node actor or ring actor.
+
+### 4.3 Local worker ring actor
+
+A local worker ring actor owns one local IPC ring lifecycle.
+
+It owns:
+
+- arena lease request and result;
+- worker ring installation;
+- worker ring readable/writable notifications;
+- ring fault and quiescence observations;
+- release proof collection;
+- arena lease release.
+
+It does not own remote endpoint routing. It can be bound to a wire edge endpoint
+by edge id, but the ring lifecycle remains local.
+
+### 4.4 GPU worker control adapter
+
+The GPU worker control adapter is the actor-facing boundary to the owned Python
+worker process.
+
+It owns or fronts:
+
+- worker process generation;
+- command serialization to the worker;
+- stdout/stderr event parsing;
+- device handle generation checks;
+- worker stop/crash/restart observations.
+
+The data plane treats this as an actor endpoint. Worker-process JSON and Python
+helper details are not exposed to MVP stage logic.
+
+### 4.5 Transport adapter actors
+
+Transport adapter actors own concrete wire byte movement.
+
+They own or front:
+
+- accepted edge streams;
+- outbound edge streams;
+- edge preamble validation;
+- byte read/write readiness;
+- transport-specific stream faults;
+- pump stop observations.
+
+The data plane treats transport events as observations on a wire edge. Transport
+actors do not decide stage readiness or object admission.
+
+### 4.6 MVP report sink
+
+The MVP report sink receives coarse data-plane outcomes and maps them to
+MVP-specific control messages.
+
+Examples:
+
+- inbound edge ready;
+- outbound edge ready;
+- object loaded for stage execution;
+- object produced for downstream transport;
+- edge faulted;
+- local edges stopped.
+
+The sink does not inspect ring cursors, arena leases, worker generations, or
+transport pump internals.
+
+---
+
+## 5. Actor API surface
+
+Concrete Rust names are schematic. The contract is the message shape and
+ownership boundary.
+
+### 5.1 Provisioning input
+
+MVP sends provisioning intent to the data-plane node actor:
+
+```text
+ProvisionDataPlaneRun {
+    run_id,
+    local_node_id,
+    arena_actor,
+    worker_actor,
+    transport_actor,
+    report_sink,
+}
+
+ProvisionWireEdgeEndpoint {
+    run_id,
+    edge_id,
+    direction,
+    edge_kind,
+    local_role_port,
+    local_node_id,
+    peer_node_id,
+    peer_endpoint,
+    object_spec,
+    transport_spec,
+    local_ring_spec,
+}
+```
+
+`direction` is relative to the local node's stage role: inbound means the local
+stage consumes objects from the edge; outbound means the local stage produces
+objects to the edge.
+
+Provisioning is declarative. MVP describes the intended edge endpoint and the
+actors available to execute effects. It does not prescribe lease/install/driver
+ordering.
+
+### 5.2 Runtime observations
+
+Runtime adapters report observations back to data-plane actors:
+
+```text
+ArenaRingLeased
+ArenaRingLeaseRejected
+ArenaRingReleased
+ArenaRingReleaseRejected
+
+WorkerReady
+WorkerRingInstalled
+WorkerRingFaulted
+WorkerRingQuiesced
+WorkerRingReadable
+WorkerRingWritable
+WorkerObjectLoaded
+WorkerObjectProduced
+WorkerObjectFailed
+WorkerStopped
+WorkerFaulted
+
+TransportEdgeReady
+TransportBytesReceived
+TransportBytesSent
+TransportStreamClosed
+TransportStreamFaulted
+TransportPumpStopped
+```
+
+Observations are facts, not commands. The data plane decides the next state and
+any follow-up messages.
+
+### 5.3 Data-plane effects
+
+The data plane sends effect requests to runtime adapter actors:
+
+```text
+LeaseArenaRing
+CancelArenaRingLease
+ReleaseArenaRingLease
+
+InstallWorkerRing
+UninstallWorkerRing
+NotifyWorkerRingReadable
+NotifyWorkerRingWritable
+LoadObjectFromWorkerRing
+ExecuteWorkerStep
+ReleaseWorkerDeviceObject
+
+EstablishWireSend
+EstablishWireRecv
+WriteWireObject
+StopWirePump
+```
+
+Effects are actor messages. The receiving adapter owns the concrete mechanism:
+memfd/mmap, JSON stdin/stdout, process supervision, iroh streams, or test doubles.
+
+### 5.4 Data-plane reports
+
+The data plane reports only stable semantic outcomes to MVP:
+
+```text
+InboundEdgeReady { edge_id }
+OutboundEdgeReady { edge_id }
+ObjectLoaded { edge_id, object_id, sequence, device_handle }
+ObjectProduced { edge_id, object_id, sequence, extent }
+EdgeFaulted { edge_id, reason }
+EdgeStopped { edge_id }
+LocalEdgesStopped { run_id }
+WorkerDataPlaneFaulted { reason }
+```
+
+Reports are the only data-plane messages MVP stage/orchestrator actors should
+need for normal stage progression.
+
+---
+
+## 6. Behavior owned by data-plane
+
+### 6.1 Edge establishment
+
+For each provisioned edge endpoint, data-plane actors own the establishment
+sequence.
+
+Inbound endpoint:
+
+```text
+provision endpoint
+  -> lease local ingress ring
+  -> install ring into worker input port
+  -> establish receive transport if the edge is remote
+  -> report inbound edge ready
+```
+
+Outbound endpoint:
+
+```text
+provision endpoint
+  -> lease local egress ring when worker output is required
+  -> install ring into worker output port
+  -> establish send transport if the edge is remote
+  -> report outbound edge ready
+```
+
+Readiness is reported only after every required local and wire resource for that
+endpoint is ready. A local-only endpoint may omit transport establishment. A
+wire-only endpoint may omit worker-ring establishment when it terminates outside
+the local GPU worker.
+
+### 6.2 Object ingress
+
+For inbound data, data-plane actors own object admission.
+
+The data plane:
+
+- associates incoming bytes with the correct wire edge and stream;
+- validates object framing and object spec constraints;
+- preserves sequence ordering required by the edge contract;
+- writes or exposes the object through the local ingress ring;
+- asks the worker to load the object to device;
+- waits for a valid worker object-loaded observation;
+- reports object loaded to MVP only after the device handle is current and the
+  logical object is complete.
+
+MVP does not parse object headers, track ingress buffers, reload cursors, or gate
+object-loaded visibility.
+
+### 6.3 Object egress
+
+For outbound data, data-plane actors own object production and forwarding.
+
+The data plane:
+
+- receives compute/output observations from the worker;
+- binds produced objects to the correct outbound edge and sequence;
+- validates object extent and object contract;
+- publishes readable/writable state to the worker and transport actors;
+- forwards complete object records on the wire when the edge is remote;
+- reports object produced or step-visible outcomes to MVP at semantic
+  boundaries, not cursor boundaries.
+
+MVP does not decide when a local output ring is readable, when a transport stream
+should consume it, or when a produced object is safe to expose downstream.
+
+### 6.4 Faults
+
+The data plane owns data movement fault classification and propagation.
+
+Fault sources include:
+
+- arena lease rejection or release rejection;
+- worker ring installation failure;
+- worker ring fault;
+- malformed object framing;
+- sequence violation;
+- worker object load/produce failure;
+- transport stream read/write/protocol failure;
+- pump stop before quiescence;
+- stale worker generation or device handle.
+
+A fault on one edge endpoint must not silently corrupt another endpoint. The data
+plane maps local faults to edge-scoped or worker-scoped reports, starts the
+required stop/quiescence path, and emits the appropriate MVP report.
+
+### 6.5 Stop, quiescence, and release
+
+The data plane owns teardown ordering for local resources.
+
+For a bound edge/ring pair, stop requires:
+
+```text
+stop transport pump if present
+  -> uninstall or quiesce worker ring if installed
+  -> prove no local reader/writer still uses the ring
+  -> release arena lease
+  -> report edge stopped
+```
+
+Arena release must be gated by quiescence proof. MVP may request run or edge
+stop, but it does not supply low-level release proof or decide when a ring is
+safe to release.
+
+---
+
+## 7. MVP-system utilization
+
+`mvp-system` remains responsible for MVP orchestration and stage semantics.
+
+It owns:
+
+- run planning and edge assignment;
+- stage provisioning authority;
+- membership/readiness gates outside the data plane;
+- weight loading and role configuration intent;
+- stage controller behavior;
+- prompt injection and token consumption;
+- mapping data-plane reports to MVP lifecycle messages;
+- selecting concrete runtime adapters for arena, worker process, and transport.
+
+For data movement, MVP code acts as a client:
+
+1. Spawn or obtain actor addresses for the data-plane node actor and required
+   runtime adapters.
+2. Send run and edge provisioning intent to the data-plane node actor.
+3. Forward runtime observations from concrete adapter actors when those adapters
+   are MVP-owned.
+4. Receive coarse data-plane reports.
+5. Translate those reports into stage-controller or orchestrator messages.
+
+MVP must not rely on private edge states such as waiting-for-lease,
+waiting-for-worker-ring, waiting-for-driver, pump-stopped, ring-quiesced, or
+release-ready. Those are data-plane implementation states.
+
+---
+
+## 8. Required invariants
+
+- `EdgeId` names a logical run-plan edge, not a local ring allocation.
+- `RingId` names a local arena-backed IPC ring, not a distributed edge.
+- A ring may be bound to an edge endpoint, but the identifiers are not
+  interchangeable.
+- Actor messages carry control and identities, not tensor payload bytes.
+- Object-loaded reports are emitted only for complete, validated objects with
+  current-generation device handles.
+- Edge-ready reports are emitted only after required wire and local IPC resources
+  are ready.
+- Arena lease release is gated by local quiescence proof.
+- Worker process generation is part of every device-handle validity decision.
+- Transport faults and worker faults are classified by data-plane before they
+  become MVP reports.
+- MVP stage logic observes semantic outcomes, never ring cursor mechanics.
+
+---
+
+## 9. Non-goals
+
+`data-plane` does not own:
+
+- global run planning;
+- placement optimization;
+- model layer assignment;
+- weight download or weight loading semantics;
+- prompt tokenization or output token policy;
+- membership convergence;
+- provider provisioning;
+- concrete iroh endpoint construction;
+- concrete Python helper implementation.
+
+The crate defines reusable actor protocols and data-movement behavior. Concrete
+runtime adapters may live beside MVP code, inside reusable support crates, or in
+tests, as long as they satisfy the actor contracts above.

crates/data-plane/src/actor.rs

@@ -0,0 +1,769 @@
+//! Swactor-facing data-plane actor API.
+//!
+//! The actor owns lifecycle decisions for logical wire edge endpoints and their
+//! bound node-local worker rings. Runtime-specific actors execute the effect
+//! messages and report observations back.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use swactor::actor::{ActorAddress, ActorInterface};
+use swactor::runtime::Ctx;
+
+use crate::edge_lifecycle as lifecycle;
+use crate::object_record;
+
+pub use lifecycle::{
+    DType, EdgeFaultReason, EdgeId, LeaseRequestId, NodeId, ObjectKind, ObjectSpec,
+    QuiescenceProof, RingDirection, RingId, RingLayout, RingLeaseRejection, RingSpec, RunId,
+};
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PortId(pub String);
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum EdgeKind {
+    TokenIn,
+    Activation,
+    TokenOut,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum EdgeEndpointDirection {
+    Inbound,
+    Outbound,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PeerEndpoint(pub String);
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum TransportBinding {
+    Remote { endpoint: Option<PeerEndpoint> },
+    LocalOnly,
+}
+
+impl TransportBinding {
+    fn requires_transport(&self) -> bool {
+        matches!(self, Self::Remote { .. })
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum WorkerRingBinding {
+    Required,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct DataPlaneRunConfig {
+    pub run_id: RunId,
+    pub local_node_id: NodeId,
+    pub arena_actor: ActorAddress,
+    pub worker_actor: ActorAddress,
+    pub transport_actor: ActorAddress,
+    pub report_sink: ActorAddress,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct WireEdgeEndpoint {
+    pub run_id: RunId,
+    pub edge_id: EdgeId,
+    pub direction: EdgeEndpointDirection,
+    pub kind: EdgeKind,
+    pub local_node_id: NodeId,
+    pub peer_node_id: Option<NodeId>,
+    pub peer_endpoint: Option<PeerEndpoint>,
+    pub local_role_port: PortId,
+    pub object_spec: ObjectSpec,
+    pub ring_spec: RingSpec,
+    pub object_record_spec: object_record::ObjectSpec,
+    pub transport: TransportBinding,
+    pub worker_ring: WorkerRingBinding,
+}
+
+impl WireEdgeEndpoint {
+    fn provision_event(&self) -> Option<lifecycle::EdgeEvent> {
+        match self.direction {
+            EdgeEndpointDirection::Inbound => {
+                Some(lifecycle::EdgeEvent::ProvisionRx(lifecycle::ProvisionRx {
+                    run_id: self.run_id,
+                    edge_id: self.edge_id,
+                    local_node_id: self.local_node_id,
+                    object_spec: self.object_spec,
+                    ring_spec: self.ring_spec,
+                }))
+            }
+            EdgeEndpointDirection::Outbound => {
+                Some(lifecycle::EdgeEvent::ProvisionTx(lifecycle::ProvisionTx {
+                    run_id: self.run_id,
+                    edge_id: self.edge_id,
+                    local_node_id: self.local_node_id,
+                    consumer_node_id: self.peer_node_id?,
+                    object_spec: self.object_spec,
+                    ring_spec: self.ring_spec,
+                }))
+            }
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum DataPlaneNodeMsg {
+    ConfigureRun(DataPlaneRunConfig),
+    ProvisionWireEdgeEndpoint(WireEdgeEndpoint),
+    Arena(ArenaObservation),
+    Worker(WorkerObservation),
+    Transport(TransportObservation),
+    StopEdge { edge_id: EdgeId },
+    StopRun { run_id: RunId },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum ArenaObservation {
+    RingLeased {
+        request_id: LeaseRequestId,
+        ring_id: RingId,
+        layout: RingLayout,
+    },
+    RingLeaseRejected {
+        request_id: LeaseRequestId,
+        reason: RingLeaseRejection,
+    },
+    RingReleased {
+        ring_id: RingId,
+    },
+    RingReleaseRejected {
+        ring_id: RingId,
+    },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum WorkerObservation {
+    WorkerReady,
+    RingInstalled {
+        edge_id: EdgeId,
+        ring_id: RingId,
+    },
+    RingFaulted {
+        edge_id: EdgeId,
+        ring_id: RingId,
+        reason: lifecycle::RingFaultReason,
+    },
+    RingQuiesced {
+        ring_id: RingId,
+    },
+    QuiescenceProven {
+        ring_id: RingId,
+    },
+    ObjectLoaded {
+        edge_id: EdgeId,
+        ring_id: RingId,
+        object_id: object_record::ObjectId,
+        sequence: u64,
+        extent: u64,
+        handle: DeviceHandle,
+    },
+    ObjectProduced {
+        edge_id: EdgeId,
+        ring_id: RingId,
+        object_id: object_record::ObjectId,
+        sequence: u64,
+        extent: u64,
+    },
+    ObjectFailed {
+        edge_id: EdgeId,
+        ring_id: RingId,
+        reason: object_record::ObjectFailureReason,
+    },
+    WorkerFaulted,
+    WorkerStopped,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum TransportObservation {
+    EdgeReady {
+        edge_id: EdgeId,
+    },
+    BytesAvailable {
+        edge_id: EdgeId,
+        stream_id: StreamId,
+        byte_count: u64,
+    },
+    StreamClosed {
+        edge_id: EdgeId,
+    },
+    StreamFaulted {
+        edge_id: EdgeId,
+        reason: lifecycle::StreamFaultReason,
+    },
+    PumpStopped {
+        edge_id: EdgeId,
+        ring_id: RingId,
+    },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum DataPlaneArenaMsg {
+    LeaseRing {
+        request_id: LeaseRequestId,
+        edge_id: EdgeId,
+        direction: RingDirection,
+        ring_spec: RingSpec,
+    },
+    CancelQueuedLease {
+        request_id: LeaseRequestId,
+        edge_id: EdgeId,
+    },
+    ReleaseRing {
+        ring_id: RingId,
+        proof: QuiescenceProof,
+    },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum DataPlaneWorkerMsg {
+    InstallRing {
+        edge_id: EdgeId,
+        ring_id: RingId,
+        direction: RingDirection,
+        layout: RingLayout,
+        object_spec: ObjectSpec,
+        ring_spec: RingSpec,
+        role_port: PortId,
+    },
+    UninstallRing {
+        edge_id: EdgeId,
+        ring_id: RingId,
+    },
+    NotifyRingReadable {
+        ring_id: RingId,
+    },
+    NotifyRingWritable {
+        ring_id: RingId,
+    },
+    LoadObjectFromRing {
+        edge_id: EdgeId,
+        ring_id: RingId,
+        object_id: object_record::ObjectId,
+        sequence: u64,
+        extent: u64,
+    },
+    ExecuteStep {
+        edge_id: EdgeId,
+        sequence: u64,
+    },
+    ReleaseDeviceObject {
+        handle: DeviceHandle,
+    },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum DataPlaneTransportMsg {
+    EstablishSend {
+        edge_id: EdgeId,
+        consumer_node_id: NodeId,
+        ring_id: RingId,
+        layout: RingLayout,
+    },
+    EstablishRecv {
+        edge_id: EdgeId,
+        ring_id: RingId,
+        layout: RingLayout,
+    },
+    WriteWireObject {
+        edge_id: EdgeId,
+        object_id: object_record::ObjectId,
+        sequence: u64,
+        extent: u64,
+    },
+    StopWirePump {
+        edge_id: EdgeId,
+        ring_id: RingId,
+    },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum DataPlaneReportMsg {
+    InboundEdgeReady {
+        edge_id: EdgeId,
+    },
+    OutboundEdgeReady {
+        edge_id: EdgeId,
+    },
+    ObjectLoaded {
+        edge_id: EdgeId,
+        object_id: object_record::ObjectId,
+        sequence: u64,
+        extent: u64,
+        handle: DeviceHandle,
+    },
+    ObjectProduced {
+        edge_id: EdgeId,
+        object_id: object_record::ObjectId,
+        sequence: u64,
+        extent: u64,
+    },
+    EdgeFaulted {
+        edge_id: EdgeId,
+        reason: EdgeFaultReason,
+    },
+    EdgeStopped {
+        edge_id: EdgeId,
+    },
+    LocalEdgesStopped {
+        run_id: RunId,
+    },
+    WorkerDataPlaneFaulted {
+        reason: WorkerDataPlaneFaultReason,
+    },
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum WorkerDataPlaneFaultReason {
+    WorkerFaulted,
+    WorkerStopped,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct WorkerGeneration(pub u64);
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct DeviceHandle {
+    pub generation: WorkerGeneration,
+    pub id: u64,
+}
+
+impl DeviceHandle {
+    pub const fn new(generation: WorkerGeneration, id: u64) -> Self {
+        Self { generation, id }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct StreamId(pub u64);
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct EdgeBinding {
+    direction: EdgeEndpointDirection,
+    role_port_index: usize,
+    transport: TransportBinding,
+}
+
+pub struct DataPlaneNodeActor {
+    run: Option<DataPlaneRunConfig>,
+    lifecycle: lifecycle::EdgeEstablisher,
+    command_cursor: usize,
+    event_cursor: usize,
+    bindings: BTreeMap<EdgeId, EdgeBinding>,
+    stopped_edges: BTreeSet<EdgeId>,
+    role_ports: Vec<PortId>,
+    rings: BTreeMap<RingId, EdgeId>,
+    pending_reports: Vec<DataPlaneReportMsg>,
+    stopping_run: Option<RunId>,
+    local_edges_stopped_reported: bool,
+}
+
+impl DataPlaneNodeActor {
+    pub fn new(local_node_id: NodeId) -> Self {
+        Self {
+            run: None,
+            lifecycle: lifecycle::EdgeEstablisher::new(local_node_id),
+            command_cursor: 0,
+            event_cursor: 0,
+            bindings: BTreeMap::new(),
+            stopped_edges: BTreeSet::new(),
+            role_ports: Vec::new(),
+            rings: BTreeMap::new(),
+            pending_reports: Vec::new(),
+            stopping_run: None,
+            local_edges_stopped_reported: false,
+        }
+    }
+
+    fn configure(&mut self, config: DataPlaneRunConfig) {
+        self.lifecycle = lifecycle::EdgeEstablisher::new(config.local_node_id);
+        self.command_cursor = 0;
+        self.event_cursor = 0;
+        self.bindings.clear();
+        self.role_ports.clear();
+        self.stopped_edges.clear();
+        self.rings.clear();
+        self.pending_reports.clear();
+        self.stopping_run = None;
+        self.local_edges_stopped_reported = false;
+        self.run = Some(config);
+    }
+
+    fn observe(&mut self, msg: DataPlaneNodeMsg) {
+        match msg {
+            DataPlaneNodeMsg::ConfigureRun(config) => self.configure(config),
+            DataPlaneNodeMsg::ProvisionWireEdgeEndpoint(endpoint) => {
+                if let Some(event) = endpoint.provision_event() {
+                    let role_port_index = self.role_ports.len();
+                    self.role_ports.push(endpoint.local_role_port.clone());
+                    self.bindings.insert(
+                        endpoint.edge_id,
+                        EdgeBinding {
+                            direction: endpoint.direction,
+                            role_port_index,
+                            transport: endpoint.transport.clone(),
+                        },
+                    );
+                    self.lifecycle.observe(event);
+                }
+            }
+            DataPlaneNodeMsg::Arena(observation) => match observation {
+                ArenaObservation::RingLeased {
+                    request_id,
+                    ring_id,
+                    layout,
+                } => self.lifecycle.observe(lifecycle::EdgeEvent::RingLeased {
+                    request_id,
+                    ring_id,
+                    layout,
+                }),
+                ArenaObservation::RingLeaseRejected { request_id, reason } => self
+                    .lifecycle
+                    .observe(lifecycle::EdgeEvent::RingLeaseRejected { request_id, reason }),
+                ArenaObservation::RingReleased { .. }
+                | ArenaObservation::RingReleaseRejected { .. } => {}
+            },
+            DataPlaneNodeMsg::Worker(observation) => match observation {
+                WorkerObservation::WorkerReady => {}
+                WorkerObservation::RingInstalled { edge_id, ring_id } => {
+                    self.rings.insert(ring_id, edge_id);
+                    self.lifecycle
+                        .observe(lifecycle::EdgeEvent::RingInstalled { edge_id, ring_id });
+                }
+                WorkerObservation::RingFaulted {
+                    edge_id,
+                    ring_id,
+                    reason,
+                } => self.lifecycle.observe(lifecycle::EdgeEvent::RingFault {
+                    edge_id,
+                    ring_id,
+                    reason,
+                }),
+                WorkerObservation::RingQuiesced { ring_id } => self
+                    .lifecycle
+                    .observe(lifecycle::EdgeEvent::RingQuiesced { ring_id }),
+                WorkerObservation::QuiescenceProven { ring_id } => self
+                    .lifecycle
+                    .observe(lifecycle::EdgeEvent::QuiescenceProven { ring_id }),
+                WorkerObservation::ObjectLoaded {
+                    edge_id,
+                    object_id,
+                    sequence,
+                    extent,
+                    handle,
+                    ..
+                } => self.report_object_loaded(edge_id, object_id, sequence, extent, handle),
+                WorkerObservation::ObjectProduced {
+                    edge_id,
+                    object_id,
+                    sequence,
+                    extent,
+                    ..
+                } => self.report_object_produced(edge_id, object_id, sequence, extent),
+                WorkerObservation::ObjectFailed {
+                    edge_id, ring_id, ..
+                } => self.lifecycle.observe(lifecycle::EdgeEvent::RingFault {
+                    edge_id,
+                    ring_id,
+                    reason: lifecycle::RingFaultReason::WorkerRejectedRing,
+                }),
+                WorkerObservation::WorkerFaulted => {
+                    self.report_worker_fault(WorkerDataPlaneFaultReason::WorkerFaulted)
+                }
+                WorkerObservation::WorkerStopped => {
+                    self.report_worker_fault(WorkerDataPlaneFaultReason::WorkerStopped)
+                }
+            },
+            DataPlaneNodeMsg::Transport(observation) => match observation {
+                TransportObservation::EdgeReady { edge_id } => self
+                    .lifecycle
+                    .observe(lifecycle::EdgeEvent::DriverEdgeReady { edge_id }),
+                TransportObservation::BytesAvailable { .. }
+                | TransportObservation::StreamClosed { .. } => {}
+                TransportObservation::StreamFaulted { edge_id, reason } => self
+                    .lifecycle
+                    .observe(lifecycle::EdgeEvent::StreamFault { edge_id, reason }),
+                TransportObservation::PumpStopped { edge_id, ring_id } => self
+                    .lifecycle
+                    .observe(lifecycle::EdgeEvent::PumpStopped { edge_id, ring_id }),
+            },
+            DataPlaneNodeMsg::StopEdge { edge_id } => self
+                .lifecycle
+                .observe(lifecycle::EdgeEvent::StopEdge { edge_id }),
+            DataPlaneNodeMsg::StopRun { run_id } => {
+                self.stopping_run = Some(run_id);
+                self.local_edges_stopped_reported = false;
+                let edge_ids = self.bindings.keys().copied().collect::<Vec<_>>();
+                for edge_id in edge_ids {
+                    self.lifecycle
+                        .observe(lifecycle::EdgeEvent::StopEdge { edge_id });
+                }
+                self.maybe_report_local_edges_stopped();
+            }
+        }
+    }
+
+    fn ring_for_edge(&self, edge_id: EdgeId) -> Option<RingId> {
+        self.rings
+            .iter()
+            .find_map(|(ring_id, seen_edge)| (*seen_edge == edge_id).then_some(*ring_id))
+    }
+
+    fn role_port_for(&self, edge_id: EdgeId) -> PortId {
+        self.bindings
+            .get(&edge_id)
+            .and_then(|binding| self.role_ports.get(binding.role_port_index))
+            .cloned()
+            .unwrap_or_else(|| PortId(String::new()))
+    }
+
+    fn report_object_loaded(
+        &mut self,
+        edge_id: EdgeId,
+        object_id: object_record::ObjectId,
+        sequence: u64,
+        extent: u64,
+        handle: DeviceHandle,
+    ) {
+        self.pending_reports.push(DataPlaneReportMsg::ObjectLoaded {
+            edge_id,
+            object_id,
+            sequence,
+            extent,
+            handle,
+        });
+    }
+
+    fn report_object_produced(
+        &mut self,
+        edge_id: EdgeId,
+        object_id: object_record::ObjectId,
+        sequence: u64,
+        extent: u64,
+    ) {
+        self.pending_reports
+            .push(DataPlaneReportMsg::ObjectProduced {
+                edge_id,
+                object_id,
+                sequence,
+                extent,
+            });
+    }
+
+    fn report_worker_fault(&mut self, reason: WorkerDataPlaneFaultReason) {
+        self.pending_reports
+            .push(DataPlaneReportMsg::WorkerDataPlaneFaulted { reason });
+    }
+
+    fn report_local_edges_stopped(&mut self, run_id: RunId) {
+        self.pending_reports
+            .push(DataPlaneReportMsg::LocalEdgesStopped { run_id });
+    }
+
+    fn maybe_report_local_edges_stopped(&mut self) {
+        let Some(run_id) = self.stopping_run else {
+            return;
+        };
+        if self.local_edges_stopped_reported {
+            return;
+        }
+        if self
+            .bindings
+            .keys()
+            .all(|edge_id| self.stopped_edges.contains(edge_id))
+        {
+            self.local_edges_stopped_reported = true;
+            self.report_local_edges_stopped(run_id);
+        }
+    }
+
+    fn drain(&mut self, ctx: &Ctx) {
+        let Some(run) = self.run.clone() else {
+            return;
+        };
+
+        loop {
+            let mut progressed = false;
+            let mut immediate_ready = Vec::new();
+
+            while self.command_cursor < self.lifecycle.commands().len() {
+                let command = self.lifecycle.commands()[self.command_cursor].clone();
+                self.command_cursor += 1;
+                progressed = true;
+                match command {
+                    lifecycle::EdgeCommand::LeaseRing {
+                        request_id,
+                        edge_id,
+                        direction,
+                        ring_spec,
+                    } => {
+                        let _ = ctx.send(
+                            run.arena_actor,
+                            DataPlaneArenaMsg::LeaseRing {
+                                request_id,
+                                edge_id,
+                                direction,
+                                ring_spec,
+                            },
+                        );
+                    }
+                    lifecycle::EdgeCommand::CancelQueuedLease {
+                        request_id,
+                        edge_id,
+                    } => {
+                        let _ = ctx.send(
+                            run.arena_actor,
+                            DataPlaneArenaMsg::CancelQueuedLease {
+                                request_id,
+                                edge_id,
+                            },
+                        );
+                    }
+                    lifecycle::EdgeCommand::InstallWorkerRing {
+                        edge_id,
+                        ring_id,
+                        direction,
+                        layout,
+                        object_spec,
+                        ring_spec,
+                    } => {
+                        let _ = ctx.send(
+                            run.worker_actor,
+                            DataPlaneWorkerMsg::InstallRing {
+                                edge_id,
+                                ring_id,
+                                direction,
+                                layout,
+                                object_spec,
+                                ring_spec,
+                                role_port: self.role_port_for(edge_id),
+                            },
+                        );
+                    }
+                    lifecycle::EdgeCommand::UninstallWorkerRing { edge_id, ring_id } => {
+                        let _ = ctx.send(
+                            run.worker_actor,
+                            DataPlaneWorkerMsg::UninstallRing { edge_id, ring_id },
+                        );
+                    }
+                    lifecycle::EdgeCommand::EstablishSend {
+                        edge_id,
+                        consumer_node_id,
+                        layout,
+                    } => {
+                        let ring_id = self.ring_for_edge(edge_id).unwrap_or(RingId(0));
+                        if self
+                            .bindings
+                            .get(&edge_id)
+                            .is_some_and(|binding| binding.transport.requires_transport())
+                        {
+                            let _ = ctx.send(
+                                run.transport_actor,
+                                DataPlaneTransportMsg::EstablishSend {
+                                    edge_id,
+                                    consumer_node_id,
+                                    ring_id,
+                                    layout,
+                                },
+                            );
+                        } else {
+                            immediate_ready.push(edge_id);
+                        }
+                    }
+                    lifecycle::EdgeCommand::EstablishRecv { edge_id, layout } => {
+                        let ring_id = self.ring_for_edge(edge_id).unwrap_or(RingId(0));
+                        if self
+                            .bindings
+                            .get(&edge_id)
+                            .is_some_and(|binding| binding.transport.requires_transport())
+                        {
+                            let _ = ctx.send(
+                                run.transport_actor,
+                                DataPlaneTransportMsg::EstablishRecv {
+                                    edge_id,
+                                    ring_id,
+                                    layout,
+                                },
+                            );
+                        } else {
+                            immediate_ready.push(edge_id);
+                        }
+                    }
+                    lifecycle::EdgeCommand::StopPump { edge_id, ring_id } => {
+                        let _ = ctx.send(
+                            run.transport_actor,
+                            DataPlaneTransportMsg::StopWirePump { edge_id, ring_id },
+                        );
+                    }
+                    lifecycle::EdgeCommand::ReleaseArenaLease { ring_id, proof } => {
+                        let _ = ctx.send(
+                            run.arena_actor,
+                            DataPlaneArenaMsg::ReleaseRing { ring_id, proof },
+                        );
+                    }
+                }
+            }
+
+            for edge_id in immediate_ready {
+                self.lifecycle
+                    .observe(lifecycle::EdgeEvent::DriverEdgeReady { edge_id });
+            }
+
+            while self.event_cursor < self.lifecycle.events().len() {
+                let event = self.lifecycle.events()[self.event_cursor].clone();
+                self.event_cursor += 1;
+                progressed = true;
+                match event {
+                    lifecycle::EdgeLifecycleEvent::EdgeReady { edge_id, .. } => {
+                        let report =
+                            match self.bindings.get(&edge_id).map(|binding| binding.direction) {
+                                Some(EdgeEndpointDirection::Inbound) => {
+                                    DataPlaneReportMsg::InboundEdgeReady { edge_id }
+                                }
+                                Some(EdgeEndpointDirection::Outbound) => {
+                                    DataPlaneReportMsg::OutboundEdgeReady { edge_id }
+                                }
+                                None => continue,
+                            };
+                        let _ = ctx.send(run.report_sink, report);
+                    }
+                    lifecycle::EdgeLifecycleEvent::EdgeFaulted { edge_id, reason } => {
+                        let _ = ctx.send(
+                            run.report_sink,
+                            DataPlaneReportMsg::EdgeFaulted { edge_id, reason },
+                        );
+                    }
+                    lifecycle::EdgeLifecycleEvent::EdgeStopped { edge_id } => {
+                        self.stopped_edges.insert(edge_id);
+                        let _ =
+                            ctx.send(run.report_sink, DataPlaneReportMsg::EdgeStopped { edge_id });
+                        self.maybe_report_local_edges_stopped();
+                    }
+                }
+            }
+
+            for report in self.pending_reports.drain(..) {
+                progressed = true;
+                let _ = ctx.send(run.report_sink, report);
+            }
+
+            if !progressed {
+                break;
+            }
+        }
+    }
+}
+
+impl ActorInterface for DataPlaneNodeActor {
+    type Incoming = DataPlaneNodeMsg;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) {
+        self.observe(msg);
+        self.drain(ctx);
+    }
+}

crates/data-plane/src/arena.rs

@@ -1,7 +1,10 @@
 //! Reusable arena-backed ring allocation contracts.
 
 use std::collections::{BTreeMap, VecDeque};
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use datastream::Record;
+use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ArenaSnapshot {
@@ -17,6 +20,44 @@ pub struct ArenaSnapshot {
     pub release_failures_total: u64,
 }
 
+pub const ARENA_SAMPLE_CHANNEL: &str = "mvp.arena";
+pub const ARENA_SAMPLE_INTERVAL: Duration = Duration::from_secs(1);
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ArenaSample {
+    pub seq: u64,
+    pub sample_unix_ms: u64,
+    pub capacity_bytes: u64,
+    pub live_bytes: u64,
+    pub free_bytes: u64,
+    pub active_leases: u64,
+    pub pending_leases: u64,
+    pub largest_free_range_bytes: u64,
+    pub allocation_failures_total: u64,
+    pub release_failures_total: u64,
+}
+
+impl Record for ArenaSample {
+    const CHANNEL: &'static str = ARENA_SAMPLE_CHANNEL;
+}
+
+impl From<ArenaSnapshot> for ArenaSample {
+    fn from(snapshot: ArenaSnapshot) -> Self {
+        Self {
+            seq: snapshot.seq,
+            sample_unix_ms: snapshot.sample_unix_ms,
+            capacity_bytes: snapshot.capacity_bytes,
+            live_bytes: snapshot.live_bytes,
+            free_bytes: snapshot.free_bytes,
+            active_leases: snapshot.active_leases,
+            pending_leases: snapshot.pending_leases,
+            largest_free_range_bytes: snapshot.largest_free_range_bytes,
+            allocation_failures_total: snapshot.allocation_failures_total,
+            release_failures_total: snapshot.release_failures_total,
+        }
+    }
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct NodeId(pub u64);
 

crates/data-plane/src/egress.rs

@@ -7,7 +7,7 @@ pub struct EdgeId(pub u64);
 #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct PortId(pub String);
 
-pub use data_plane::object_record::{
+pub use crate::object_record::{
     FLAG_BEGIN_SEQUENCE, FLAG_END_OF_SEQUENCE, HEADER_LEN, KNOWN_FLAGS_MASK, OBJECT_MAGIC,
     OBJECT_MAGIC_BYTES, OBJECT_VERSION, ObjectFlags, ObjectHeader, ObjectId, ObjectLayout,
     ObjectSpec,
@@ -127,7 +127,6 @@ pub enum WakeHint {
     RingWritable { ring_id: RingId },
 }
 
-#[cfg(test)]
 #[derive(Clone, Debug, PartialEq, Eq)]
 struct PendingStep {
     step_id: StepId,
@@ -135,7 +134,6 @@ struct PendingStep {
     role_state_updated: bool,
 }
 
-#[cfg(test)]
 pub struct EgressProducerHarness {
     generation: WorkerGeneration,
     rings: std::collections::BTreeMap<RingId, InstallRing>,
@@ -149,7 +147,6 @@ pub struct EgressProducerHarness {
     events: Vec<WorkerEgressOut>,
 }
 
-#[cfg(test)]
 impl EgressProducerHarness {
     pub fn new(generation: WorkerGeneration) -> Self {
         Self {
@@ -369,7 +366,6 @@ impl EgressProducerHarness {
     }
 }
 
-#[cfg(test)]
 fn encode_header(output: OutputBinding) -> Vec<u8> {
     ObjectHeader {
         object_id: output.object_id,

crates/data-plane/src/ingress.rs

@@ -7,7 +7,7 @@ pub struct EdgeId(pub u64);
 #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct PortId(pub String);
 
-pub use data_plane::object_record::{
+pub use crate::object_record::{
     FLAG_BEGIN_SEQUENCE, FLAG_END_OF_SEQUENCE, HEADER_LEN, KNOWN_FLAGS_MASK, OBJECT_MAGIC,
     OBJECT_MAGIC_BYTES, OBJECT_VERSION, ObjectFailureReason, ObjectFlags, ObjectHeader, ObjectId,
     ObjectLayout, ObjectRecord, ObjectRecordBuilder, ObjectRecordRead, ObjectSpec,
@@ -106,7 +106,6 @@ pub struct DeviceCopyLog {
     pub byte_count: u64,
 }
 
-#[cfg(test)]
 #[derive(Clone, Debug, PartialEq, Eq)]
 struct PendingObject {
     record: ObjectRecord,
@@ -114,7 +113,6 @@ struct PendingObject {
     handle: Option<DeviceHandle>,
 }
 
-#[cfg(test)]
 pub struct IngressParserHarness {
     generation: WorkerGeneration,
     install: Option<InstallRing>,
@@ -128,7 +126,6 @@ pub struct IngressParserHarness {
     events: Vec<WorkerIngressOut>,
 }
 
-#[cfg(test)]
 impl IngressParserHarness {
     pub fn new(generation: WorkerGeneration) -> Self {
         Self {
@@ -300,7 +297,6 @@ impl IngressParserHarness {
     }
 }
 
-#[cfg(test)]
 fn object_failure_metadata(
     bytes: &[u8],
     reason: ObjectFailureReason,

crates/data-plane/src/lib.rs

@@ -1,5 +1,11 @@
-//! Reusable local data-plane contracts for arena-backed byte movement.
+//! Reusable actor-oriented data-plane contracts for wire edges, local IPC rings,
+//! arena-backed byte movement, and GPU worker object movement.
 
+pub mod actor;
 pub mod arena;
+pub mod edge_actor;
+pub mod edge_lifecycle;
+pub mod egress;
+pub mod ingress;
 pub mod object_record;
 pub mod ring;

crates/data-plane/tests/data_plane_actor_guarantees.rs

@@ -0,0 +1,231 @@
+use data_plane::actor as dp;
+use data_plane::object_record;
+use swactor::config::RuntimeConfig;
+use swactor::runtime::Runtime;
+
+fn object_spec() -> dp::ObjectSpec {
+    dp::ObjectSpec {
+        kind: dp::ObjectKind::Activation,
+        dtype: dp::DType::F16,
+        max_extent_bytes: 4096,
+    }
+}
+
+fn record_spec() -> object_record::ObjectSpec {
+    object_record::ObjectSpec {
+        max_extent: 4096,
+        alignment: 4,
+        layout: object_record::ObjectLayout::Token,
+    }
+}
+
+fn ring_spec() -> dp::RingSpec {
+    dp::RingSpec {
+        header_bytes: 128,
+        data_bytes: 4096,
+        alignment: 64,
+    }
+}
+
+fn layout() -> dp::RingLayout {
+    dp::RingLayout {
+        start_offset: 0,
+        header_offset: 0,
+        data_offset: 128,
+        end_offset: 4224,
+        data_bytes: 4096,
+        alignment: 64,
+    }
+}
+
+fn inbound_endpoint() -> dp::WireEdgeEndpoint {
+    dp::WireEdgeEndpoint {
+        run_id: dp::RunId(55),
+        edge_id: dp::EdgeId(7001),
+        direction: dp::EdgeEndpointDirection::Inbound,
+        kind: dp::EdgeKind::Activation,
+        local_node_id: dp::NodeId(10),
+        peer_node_id: Some(dp::NodeId(9)),
+        peer_endpoint: Some(dp::PeerEndpoint("node-9".to_owned())),
+        local_role_port: dp::PortId("input".to_owned()),
+        object_spec: object_spec(),
+        ring_spec: ring_spec(),
+        object_record_spec: record_spec(),
+        transport: dp::TransportBinding::Remote {
+            endpoint: Some(dp::PeerEndpoint("node-9".to_owned())),
+        },
+        worker_ring: dp::WorkerRingBinding::Required,
+    }
+}
+
+#[test]
+fn inbound_wire_edge_establishes_through_arena_worker_transport_then_reports_ready() {
+    let runtime = Runtime::new(RuntimeConfig::default());
+    let arena = runtime
+        .new_inbox::<dp::DataPlaneArenaMsg>()
+        .expect("arena inbox");
+    let worker = runtime
+        .new_inbox::<dp::DataPlaneWorkerMsg>()
+        .expect("worker inbox");
+    let transport = runtime
+        .new_inbox::<dp::DataPlaneTransportMsg>()
+        .expect("transport inbox");
+    let reports = runtime
+        .new_inbox::<dp::DataPlaneReportMsg>()
+        .expect("report inbox");
+    let actor = runtime
+        .spawn(dp::DataPlaneNodeActor::new(dp::NodeId(10)))
+        .expect("spawn data-plane actor");
+
+    runtime
+        .send_to(
+            actor,
+            dp::DataPlaneNodeMsg::ConfigureRun(dp::DataPlaneRunConfig {
+                run_id: dp::RunId(55),
+                local_node_id: dp::NodeId(10),
+                arena_actor: *arena.addr(),
+                worker_actor: *worker.addr(),
+                transport_actor: *transport.addr(),
+                report_sink: *reports.addr(),
+            }),
+        )
+        .expect("configure run");
+    runtime
+        .send_to(
+            actor,
+            dp::DataPlaneNodeMsg::ProvisionWireEdgeEndpoint(inbound_endpoint()),
+        )
+        .expect("provision inbound");
+    runtime.tick();
+
+    assert_eq!(
+        arena.try_recv(),
+        Some(dp::DataPlaneArenaMsg::LeaseRing {
+            request_id: dp::LeaseRequestId(1),
+            edge_id: dp::EdgeId(7001),
+            direction: dp::RingDirection::Ingress,
+            ring_spec: ring_spec(),
+        })
+    );
+
+    runtime
+        .send_to(
+            actor,
+            dp::DataPlaneNodeMsg::Arena(dp::ArenaObservation::RingLeased {
+                request_id: dp::LeaseRequestId(1),
+                ring_id: dp::RingId(8001),
+                layout: layout(),
+            }),
+        )
+        .expect("ring leased");
+    runtime.tick();
+
+    assert_eq!(
+        worker.try_recv(),
+        Some(dp::DataPlaneWorkerMsg::InstallRing {
+            edge_id: dp::EdgeId(7001),
+            ring_id: dp::RingId(8001),
+            direction: dp::RingDirection::Ingress,
+            layout: layout(),
+            object_spec: object_spec(),
+            ring_spec: ring_spec(),
+            role_port: dp::PortId("input".to_owned()),
+        })
+    );
+
+    runtime
+        .send_to(
+            actor,
+            dp::DataPlaneNodeMsg::Worker(dp::WorkerObservation::RingInstalled {
+                edge_id: dp::EdgeId(7001),
+                ring_id: dp::RingId(8001),
+            }),
+        )
+        .expect("worker installed");
+    runtime.tick();
+
+    assert_eq!(
+        transport.try_recv(),
+        Some(dp::DataPlaneTransportMsg::EstablishRecv {
+            edge_id: dp::EdgeId(7001),
+            ring_id: dp::RingId(8001),
+            layout: layout(),
+        })
+    );
+
+    runtime
+        .send_to(
+            actor,
+            dp::DataPlaneNodeMsg::Transport(dp::TransportObservation::EdgeReady {
+                edge_id: dp::EdgeId(7001),
+            }),
+        )
+        .expect("transport ready");
+    runtime.tick();
+
+    assert_eq!(
+        reports.try_recv(),
+        Some(dp::DataPlaneReportMsg::InboundEdgeReady {
+            edge_id: dp::EdgeId(7001)
+        })
+    );
+}
+
+#[test]
+fn object_loaded_observation_is_reported_as_coarse_data_plane_outcome() {
+    let runtime = Runtime::new(RuntimeConfig::default());
+    let arena = runtime
+        .new_inbox::<dp::DataPlaneArenaMsg>()
+        .expect("arena inbox");
+    let worker = runtime
+        .new_inbox::<dp::DataPlaneWorkerMsg>()
+        .expect("worker inbox");
+    let transport = runtime
+        .new_inbox::<dp::DataPlaneTransportMsg>()
+        .expect("transport inbox");
+    let reports = runtime
+        .new_inbox::<dp::DataPlaneReportMsg>()
+        .expect("report inbox");
+    let actor = runtime
+        .spawn(dp::DataPlaneNodeActor::new(dp::NodeId(10)))
+        .expect("spawn data-plane actor");
+
+    runtime
+        .send_to(
+            actor,
+            dp::DataPlaneNodeMsg::ConfigureRun(dp::DataPlaneRunConfig {
+                run_id: dp::RunId(55),
+                local_node_id: dp::NodeId(10),
+                arena_actor: *arena.addr(),
+                worker_actor: *worker.addr(),
+                transport_actor: *transport.addr(),
+                report_sink: *reports.addr(),
+            }),
+        )
+        .expect("configure run");
+    runtime
+        .send_to(
+            actor,
+            dp::DataPlaneNodeMsg::Worker(dp::WorkerObservation::ObjectLoaded {
+                edge_id: dp::EdgeId(7001),
+                ring_id: dp::RingId(8001),
+                object_id: object_record::ObjectId(9000),
+                sequence: 7,
+                extent: 16,
+                handle: dp::DeviceHandle::new(dp::WorkerGeneration(3), 42),
+            }),
+        )
+        .expect("object loaded");
+    runtime.tick();
+
+    assert_eq!(
+        reports.try_recv(),
+        Some(dp::DataPlaneReportMsg::ObjectLoaded {
+            edge_id: dp::EdgeId(7001),
+            object_id: object_record::ObjectId(9000),
+            sequence: 7,
+            extent: 16,
+            handle: dp::DeviceHandle::new(dp::WorkerGeneration(3), 42),
+        })
+    );
+}

crates/data-plane/tests/edge_lifecycle_guarantees.rs

@@ -1,4 +1,4 @@
-//! Black-box contract tests for MVP edge establishment.
+//! Black-box contract tests for data-plane edge establishment.
 //!
 //! These tests intentionally know only the public EdgeEstablisher surface:
 //!
@@ -7,9 +7,9 @@
 //! - arena, worker/token, driver, ready, fault, and release commands out
 //!
 //! They assert the guarantees in
-//! `specs/mvp_system/edge_establisher_contract.md`.
+//! the reusable data-plane edge lifecycle contract.
 
-use mvp_system::node::edge_lifecycle as edge;
+use data_plane::edge_lifecycle as edge;
 
 // A send provision carries the consumer node id because the driver must know
 // where to send. It deliberately carries no remote actor address.

crates/data-plane/tests/egress_guarantees.rs

@@ -1,16 +1,15 @@
-//! Black-box contract tests for MVP GPU worker egress production.
+//! Black-box contract tests for data-plane GPU worker egress production.
 //!
-//! These tests intentionally know only the public worker egress surface:
+//! These tests intentionally know only the public data-plane egress surface:
 //!
 //! - `InstallRing`, `ExecuteStep` output bindings, device-copy outcomes,
 //!   backpressure, and shutdown events in
 //! - committed ring bytes, cursor publication, `ObjectProduced`,
 //!   `StepCompleted`, and step failures out
 //!
-//! They assert the guarantees in
+//! They assert the reusable data-plane egress producer contract.
-//! `specs/mvp_system/gpu_worker_egress_producer_contract.md`.
 
-use mvp_system::node_data::egress;
+use data_plane::egress;
 
 // A valid egress ring config supplies the edge object spec and current worker
 // generation. The producer remains free to choose copy scheduling internally.

crates/mvp-system/src/lib.rs

@@ -5,7 +5,6 @@ extern crate self as mvp_system;
 
 pub mod chat;
 pub mod node;
-pub mod node_data;
 pub mod observability;
 pub mod orchestration;
 pub mod prompt;

crates/mvp-system/src/node/data_plane_bridge.rs

@@ -0,0 +1,69 @@
+//! MVP adapter for coarse reusable data-plane actor reports.
+
+use data_plane::actor as dp;
+use swactor::actor::{ActorAddress, ActorInterface};
+use swactor::runtime::Ctx;
+
+use crate::node::actor::NodeAgentMsg;
+
+pub struct MvpDataPlaneReportSinkActor {
+    node_agent: ActorAddress,
+}
+
+impl MvpDataPlaneReportSinkActor {
+    pub fn new(node_agent: ActorAddress) -> Self {
+        Self { node_agent }
+    }
+}
+
+impl ActorInterface for MvpDataPlaneReportSinkActor {
+    type Incoming = dp::DataPlaneReportMsg;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) {
+        match msg {
+            dp::DataPlaneReportMsg::InboundEdgeReady { edge_id } => {
+                let _ = ctx.send(
+                    self.node_agent,
+                    NodeAgentMsg::MarkInboundEdgeReady { edge_id: edge_id.0 },
+                );
+            }
+            dp::DataPlaneReportMsg::OutboundEdgeReady { edge_id } => {
+                let _ = ctx.send(
+                    self.node_agent,
+                    NodeAgentMsg::MarkOutboundEdgeReady { edge_id: edge_id.0 },
+                );
+            }
+            dp::DataPlaneReportMsg::ObjectLoaded {
+                edge_id,
+                object_id,
+                sequence,
+                handle,
+                ..
+            } => {
+                let _ = ctx.send(
+                    self.node_agent,
+                    NodeAgentMsg::ObjectLoaded {
+                        edge_id: edge_id.0,
+                        object_id: object_id.0,
+                        sequence,
+                        handle_generation: handle.generation.0,
+                        handle_id: handle.id,
+                    },
+                );
+            }
+            dp::DataPlaneReportMsg::ObjectProduced { .. } => {}
+            dp::DataPlaneReportMsg::EdgeFaulted { .. }
+            | dp::DataPlaneReportMsg::WorkerDataPlaneFaulted { .. } => {
+                let _ = ctx.send(self.node_agent, NodeAgentMsg::WorkerCrashed);
+            }
+            dp::DataPlaneReportMsg::EdgeStopped { .. } => {}
+            dp::DataPlaneReportMsg::LocalEdgesStopped { run_id } => {
+                let _ = ctx.send(
+                    self.node_agent,
+                    NodeAgentMsg::LocalEdgesStopped { run_id: run_id.0 },
+                );
+            }
+        }
+    }
+}

crates/mvp-system/src/node/mod.rs

@@ -5,5 +5,5 @@
 
 pub mod actor;
 pub mod boot_lifecycle;
-pub mod edge_lifecycle;
+pub mod data_plane_bridge;
 pub mod worker_node_runtime;

crates/mvp-system/src/node/worker_node_runtime.rs

@@ -26,9 +26,6 @@ use crate::node::actor::{
     NodeAgentActor, NodeAgentMsg, NodeAgentReport, StageCommandWire, StageInboundEdgeWire,
     StageObjectSpecWire, StageOutboundEdgeWire, StageRingSpecWire,
 };
-use crate::node::edge_lifecycle as edge;
-use crate::node_data::arena;
-use crate::node_data::ingress;
 use crate::observability::benchmark;
 use crate::orchestration::distribution_stack::DistributionRuntimeStack;
 use crate::orchestration::provider_adapters::relay::relay_runtime_config_from_env;
@@ -41,6 +38,9 @@ use crate::transport::driver_pumps as driver_model;
 use crate::transport::endpoint_advertisement::{
     EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint,
 };
+use data_plane::arena;
+use data_plane::edge_lifecycle as edge;
+use data_plane::ingress;
 use distribution::node::DistributedNodeConfig;
 use distribution::swim::telemetry::ObservedProbeEvent;
 use distribution::telemetry::{MembershipTransition, SwimProbeEvent};
@@ -737,7 +737,7 @@ fn spawn_arena_sampler(
         loop {
             interval.tick().await;
 
-            let sample = arena_manager.lock().sample(seq);
+            let sample: arena::ArenaSample = arena_manager.lock().sample(seq).into();
             seq = seq.saturating_add(1);
             producer.submit_record(channel, &sample);
         }

crates/mvp-system/src/node_data/arena.rs

@@ -1,131 +0,0 @@
-//! MVP arena-manager adapter over reusable data-plane arena contracts.
-
-use std::time::Duration;
-
-use datastream::Record;
-use serde::{Deserialize, Serialize};
-
-pub use data_plane::arena::{
-    ArenaCommand, ArenaConfig, ArenaEvent, ArenaFault, ArenaRequest, LayoutPointer, LeaseRequestId,
-    LeaseRing, NodeId, QuiescenceProof, RingId, RingLayout, RingLease, RingLeaseRejection,
-    RingReleaseRejection, RingSpec,
-};
-
-pub const ARENA_SAMPLE_CHANNEL: &str = "mvp.arena";
-pub const ARENA_SAMPLE_INTERVAL: Duration = Duration::from_secs(1);
-
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct ArenaSample {
-    pub seq: u64,
-    pub sample_unix_ms: u64,
-    pub capacity_bytes: u64,
-    pub live_bytes: u64,
-    pub free_bytes: u64,
-    pub active_leases: u64,
-    pub pending_leases: u64,
-    pub largest_free_range_bytes: u64,
-    pub allocation_failures_total: u64,
-    pub release_failures_total: u64,
-}
-
-impl Record for ArenaSample {
-    const CHANNEL: &'static str = ARENA_SAMPLE_CHANNEL;
-}
-
-impl From<data_plane::arena::ArenaSnapshot> for ArenaSample {
-    fn from(snapshot: data_plane::arena::ArenaSnapshot) -> Self {
-        Self {
-            seq: snapshot.seq,
-            sample_unix_ms: snapshot.sample_unix_ms,
-            capacity_bytes: snapshot.capacity_bytes,
-            live_bytes: snapshot.live_bytes,
-            free_bytes: snapshot.free_bytes,
-            active_leases: snapshot.active_leases,
-            pending_leases: snapshot.pending_leases,
-            largest_free_range_bytes: snapshot.largest_free_range_bytes,
-            allocation_failures_total: snapshot.allocation_failures_total,
-            release_failures_total: snapshot.release_failures_total,
-        }
-    }
-}
-
-pub struct ArenaManager {
-    inner: data_plane::arena::ArenaManager,
-}
-
-impl ArenaManager {
-    pub fn boot(config: ArenaConfig) -> Result<Self, ArenaFault> {
-        data_plane::arena::ArenaManager::boot(config).map(|inner| Self { inner })
-    }
-
-    pub fn request(&mut self, request: ArenaRequest) -> Vec<ArenaEvent> {
-        self.inner.request(request)
-    }
-
-    pub fn live_leases(&self) -> &[RingLease] {
-        self.inner.live_leases()
-    }
-
-    pub fn lookup_lease(&self, ring_id: RingId) -> Option<&RingLease> {
-        self.inner.lookup_lease(ring_id)
-    }
-
-    pub fn sample(&self, seq: u64) -> ArenaSample {
-        self.inner.sample(seq).into()
-    }
-
-    #[cfg(target_os = "linux")]
-    pub fn arena_fd(&self) -> std::os::fd::RawFd {
-        self.inner.arena_fd()
-    }
-
-    #[cfg(target_os = "linux")]
-    pub fn write_arena(&self, offset: u64, bytes: &[u8]) -> Result<(), std::io::Error> {
-        self.inner.write_arena(offset, bytes)
-    }
-
-    #[cfg(target_os = "linux")]
-    pub fn read_arena(&self, offset: u64, len: usize) -> Result<Vec<u8>, std::io::Error> {
-        self.inner.read_arena(offset, len)
-    }
-}
-
-pub struct ArenaManagerHarness {
-    manager: ArenaManager,
-    events: Vec<ArenaEvent>,
-    commands: Vec<ArenaCommand>,
-}
-
-impl ArenaManagerHarness {
-    pub fn boot(config: ArenaConfig) -> Result<Self, ArenaFault> {
-        Ok(Self {
-            manager: ArenaManager::boot(config)?,
-            events: Vec::new(),
-            commands: Vec::new(),
-        })
-    }
-
-    pub fn request(&mut self, request: ArenaRequest) {
-        self.events.extend(self.manager.request(request));
-    }
-
-    pub fn events(&self) -> &[ArenaEvent] {
-        &self.events
-    }
-
-    pub fn commands(&self) -> &[ArenaCommand] {
-        &self.commands
-    }
-
-    pub fn live_leases(&self) -> &[RingLease] {
-        self.manager.live_leases()
-    }
-
-    pub fn lookup_lease(&self, ring_id: RingId) -> Option<&RingLease> {
-        self.manager.lookup_lease(ring_id)
-    }
-
-    pub fn sample(&self, seq: u64) -> ArenaSample {
-        self.manager.sample(seq)
-    }
-}

crates/mvp-system/src/node_data/mod.rs

@@ -1,25 +0,0 @@
-//! MVP node-local data-plane adapter public surface.
-//!
-//! Reusable arena, ring, and object-record contracts live in `data-plane`.
-//! This module binds those contracts to MVP worker ingress, worker egress,
-//! and edge actor behavior.
-
-pub mod arena;
-pub mod edge_actor;
-pub mod ingress;
-
-pub mod ring {
-    pub use data_plane::ring::*;
-}
-
-pub mod object {
-    pub use data_plane::object_record::*;
-}
-
-pub mod egress {
-    pub use crate::worker::egress::*;
-}
-
-pub mod reusable {
-    pub use data_plane::{arena, object_record, ring};
-}

crates/mvp-system/src/observability/telemetry.rs

@@ -4,9 +4,9 @@ use datastream::hardware::net::HostNetSample;
 use datastream::{ChannelRegistry, Record};
 use serde::{Deserialize, Serialize};
 
-use crate::node_data::arena::ArenaSample;
 use crate::observability::lifecycle as obs;
 use crate::orchestration::provisioning::{self, ProvisionLogStream};
+use data_plane::arena::ArenaSample;
 
 /// Structured MVP lifecycle facts: run, node, stage, edge, ring, object, step, and worker events.
 pub const MVP_LIFECYCLE: &str = "mvp.lifecycle";

crates/mvp-system/src/orchestration/app.rs

@@ -22,7 +22,6 @@ use crate::orchestration::config::{DEFAULT_CONFIG_PATH, TomlConfigOverlay};
 use crate::transport::codec_registry::register_mvp_actor_codecs;
 const PROVIDER_START_MAX_ATTEMPTS: usize = 4;
 
-use crate::node_data::object as ingress;
 use crate::observability::telemetry::{
     MVP_PROVISIONING_EVENTS, MvpProvisionEventRecord, MvpProvisionLogRecord,
     mvp_provision_log_channel,
@@ -53,6 +52,7 @@ use crate::staging::gguf_shard::{StageShardPlan, plan_stage_shard};
 use crate::transport::endpoint_advertisement::{
     EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint,
 };
+use data_plane::object_record as ingress;
 use datastream::{
     ChannelContent, ChannelId, ChannelRef, DatastreamEndpoint, DatastreamEvent, DatastreamProducer,
     DatastreamPublisherMsg, DatastreamSubscribe, Frame, Lifetime, NodeId, Record, StreamDescriptor,

crates/mvp-system/src/tests/arena_manager_guarantees.rs

@@ -1,416 +0,0 @@
-//! Black-box contract tests for MVP ArenaManager behavior.
-//!
-//! These tests intentionally know only the public arena-manager surface:
-//!
-//! - arena boot configuration
-//! - lease, cancel, release, quiescence, and shutdown requests in
-//! - lease events, rejected requests, released ranges, and faults out
-//!
-//! They assert the guarantees in
-//! `specs/mvp_system/arena_manager_contract.md`.
-
-use mvp_system::node_data::arena;
-
-// A small deterministic arena makes overlap, alignment, queueing, and reuse
-// proofs easy to inspect. The concrete mmap strategy remains outside the test.
-fn arena_config() -> arena::ArenaConfig {
-    arena::ArenaConfig {
-        node_id: arena::NodeId(10),
-        reservation_ceiling: 4096,
-        base_alignment: 64,
-    }
-}
-
-// Lease requests name size and alignment only. They do not request pointers or
-// private allocator slots, which keeps layout authority inside ArenaManager.
-fn lease_request(request_id: u64, bytes: u64, alignment: u64) -> arena::LeaseRing {
-    arena::LeaseRing {
-        request_id: arena::LeaseRequestId(request_id),
-        ring_spec: arena::RingSpec {
-            header_bytes: 128,
-            data_bytes: bytes,
-            alignment,
-        },
-    }
-}
-
-// The harness exposes public lease events and lease snapshots. Tests use those
-// snapshots only after a RingLeased event, so private allocator state remains
-// unobservable.
-fn new_arena() -> arena::ArenaManagerHarness {
-    arena::ArenaManagerHarness::boot(arena_config()).expect("test arena must boot")
-}
-
-// This helper proves two public layouts are disjoint using half-open ranges.
-// It is more useful than comparing offsets directly because allocator choice is
-// intentionally implementation-defined.
-fn assert_non_overlapping(left: &arena::RingLayout, right: &arena::RingLayout) {
-    let left_range = left.start_offset..left.end_offset;
-    let right_range = right.start_offset..right.end_offset;
-    assert!(
-        left_range.end <= right_range.start || right_range.end <= left_range.start,
-        "live leases overlap: {left:?} and {right:?}"
-    );
-}
-
-// This proves sampling an unused arena reports only reserved capacity and no
-// allocator activity.
-#[test]
-fn sample_reports_empty_arena_capacity_and_zero_activity() {
-    let harness = new_arena();
-
-    let sample = harness.sample(41);
-
-    assert_eq!(sample.seq, 41);
-    assert!(
-        sample.sample_unix_ms > 0,
-        "sample timestamp must be a populated Unix epoch millisecond"
-    );
-    assert_eq!(sample.capacity_bytes, arena_config().reservation_ceiling);
-    assert_eq!(sample.live_bytes, 0);
-    assert_eq!(sample.free_bytes, arena_config().reservation_ceiling);
-    assert_eq!(sample.active_leases, 0);
-    assert_eq!(sample.pending_leases, 0);
-    assert_eq!(
-        sample.largest_free_range_bytes,
-        arena_config().reservation_ceiling
-    );
-    assert_eq!(sample.allocation_failures_total, 0);
-    assert_eq!(sample.release_failures_total, 0);
-}
-
-// This proves a granted lease is reflected in live capacity accounting and the
-// active lease count without relying on private allocator slots.
-#[test]
-fn sample_counts_live_bytes_and_active_leases_after_grant() {
-    let mut harness = new_arena();
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
-    let lease = harness.live_leases()[0].clone();
-    let lease_bytes = lease.layout.end_offset - lease.layout.start_offset;
-
-    let sample = harness.sample(42);
-
-    assert_eq!(sample.seq, 42);
-    assert_eq!(sample.capacity_bytes, arena_config().reservation_ceiling);
-    assert_eq!(sample.live_bytes, lease_bytes);
-    assert_eq!(
-        sample.free_bytes,
-        arena_config().reservation_ceiling - lease_bytes
-    );
-    assert_eq!(sample.active_leases, 1);
-    assert_eq!(sample.pending_leases, 0);
-    assert_eq!(sample.allocation_failures_total, 0);
-    assert_eq!(sample.release_failures_total, 0);
-}
-
-// This proves a proof-backed release returns the full range to the free pool and
-// restores the largest allocatable span.
-#[test]
-fn sample_reports_full_free_space_after_release() {
-    let mut harness = new_arena();
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
-    let ring_id = harness.live_leases()[0].ring_id;
-
-    harness.request(arena::ArenaRequest::ReleaseRing {
-        ring_id,
-        proof: arena::QuiescenceProof::verified(),
-    });
-    let sample = harness.sample(43);
-
-    assert_eq!(sample.live_bytes, 0);
-    assert_eq!(sample.free_bytes, arena_config().reservation_ceiling);
-    assert_eq!(sample.active_leases, 0);
-    assert_eq!(sample.pending_leases, 0);
-    assert_eq!(
-        sample.largest_free_range_bytes,
-        arena_config().reservation_ceiling
-    );
-}
-
-// This proves queued leases are visible as pending work while existing live
-// leases continue to own their bytes.
-#[test]
-fn sample_counts_queued_requests_as_pending_leases() {
-    let mut harness = new_arena();
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 3072, 64)));
-    let live = harness.live_leases()[0].clone();
-    let live_bytes = live.layout.end_offset - live.layout.start_offset;
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64)));
-
-    let sample = harness.sample(44);
-
-    assert_eq!(sample.live_bytes, live_bytes);
-    assert_eq!(
-        sample.free_bytes,
-        arena_config().reservation_ceiling - live_bytes
-    );
-    assert_eq!(sample.active_leases, 1);
-    assert_eq!(sample.pending_leases, 1);
-    assert_eq!(sample.allocation_failures_total, 0);
-    assert_eq!(sample.release_failures_total, 0);
-}
-
-// This proves lease rejection increments the allocation failure counter without
-// changing the arena's free capacity.
-#[test]
-fn sample_counts_rejected_leases_as_allocation_failures() {
-    let mut harness = new_arena();
-
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 8192, 64)));
-    let sample = harness.sample(45);
-
-    assert_eq!(sample.live_bytes, 0);
-    assert_eq!(sample.free_bytes, arena_config().reservation_ceiling);
-    assert_eq!(sample.active_leases, 0);
-    assert_eq!(sample.pending_leases, 0);
-    assert_eq!(
-        sample.largest_free_range_bytes,
-        arena_config().reservation_ceiling
-    );
-    assert_eq!(sample.allocation_failures_total, 1);
-    assert_eq!(sample.release_failures_total, 0);
-}
-
-// This proves release rejection increments the release failure counter and keeps
-// the live lease accounted as active.
-#[test]
-fn sample_counts_rejected_releases_as_release_failures() {
-    let mut harness = new_arena();
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
-    let lease = harness.live_leases()[0].clone();
-    let lease_bytes = lease.layout.end_offset - lease.layout.start_offset;
-
-    harness.request(arena::ArenaRequest::ReleaseRing {
-        ring_id: lease.ring_id,
-        proof: arena::QuiescenceProof::missing(),
-    });
-    let sample = harness.sample(46);
-
-    assert_eq!(sample.live_bytes, lease_bytes);
-    assert_eq!(
-        sample.free_bytes,
-        arena_config().reservation_ceiling - lease_bytes
-    );
-    assert_eq!(sample.active_leases, 1);
-    assert_eq!(sample.pending_leases, 0);
-    assert_eq!(sample.allocation_failures_total, 0);
-    assert_eq!(sample.release_failures_total, 1);
-}
-
-// This proves arena boot creates one stable sparse arena with one reservation
-// ceiling, offset-only layouts, and typed boot failure.
-#[test]
-fn arena_boot_creates_stable_offset_only_layout_domain() {
-    // Boot a valid arena.
-    let mut harness = new_arena();
-
-    // Lease one ring so the public layout can be inspected.
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
-    let lease = harness
-        .events()
-        .iter()
-        .find_map(|event| match event {
-            arena::ArenaEvent::RingLeased { lease } => Some(lease),
-            _ => None,
-        })
-        .expect("valid lease must be granted");
-
-    // Layout facts are arena offsets and stay under the reservation ceiling.
-    assert!(lease.layout.start_offset < arena_config().reservation_ceiling);
-    assert!(lease.layout.end_offset <= arena_config().reservation_ceiling);
-    assert!(matches!(
-        lease.layout.pointer,
-        arena::LayoutPointer::NoProcessPointer
-    ));
-
-    // Boot failure is typed and emits no usable arena.
-    let failed = arena::ArenaManagerHarness::boot(arena::ArenaConfig {
-        reservation_ceiling: 0,
-        ..arena_config()
-    });
-    assert!(matches!(
-        failed,
-        Err(arena::ArenaFault::InvalidReservationCeiling)
-    ));
-}
-
-// This proves LeaseRing either leases, queues, or rejects, and that live RingId
-// values are unique for the node lifetime.
-#[test]
-fn lease_requests_grant_queue_or_reject_with_unique_ring_ids() {
-    // Fill most of the arena with one live lease.
-    let mut harness = new_arena();
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 3072, 64)));
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64)));
-
-    // A satisfiable request under pressure may queue instead of rejecting.
-    assert!(harness.events().iter().any(|event| {
-        matches!(
-            event,
-            arena::ArenaEvent::RingLeaseQueued {
-                request_id: arena::LeaseRequestId(2)
-            }
-        )
-    }));
-
-    // A request larger than the reservation ceiling must reject.
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(3, 8192, 64)));
-    assert!(harness.events().iter().any(|event| {
-        matches!(
-            event,
-            arena::ArenaEvent::RingLeaseRejected {
-                request_id: arena::LeaseRequestId(3),
-                reason: arena::RingLeaseRejection::CannotFitWithinCeiling,
-            }
-        )
-    }));
-
-    // Granted RingId values must be unique among all live leases.
-    let ids = harness
-        .live_leases()
-        .iter()
-        .map(|lease| lease.ring_id)
-        .collect::<std::collections::BTreeSet<_>>();
-    assert_eq!(ids.len(), harness.live_leases().len());
-}
-
-// This proves live leases are non-overlapping, in-bounds, aligned, and stable
-// for the lease lifetime.
-#[test]
-fn live_layouts_are_non_overlapping_aligned_in_bounds_and_stable() {
-    // Lease two rings with explicit alignment requirements.
-    let mut harness = new_arena();
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 512, 128)));
-
-    // Read the public lease snapshots.
-    let leases = harness.live_leases().to_vec();
-    assert_eq!(leases.len(), 2);
-
-    // Prove non-overlap and in-bounds without constraining allocator placement.
-    assert_non_overlapping(&leases[0].layout, &leases[1].layout);
-    for lease in &leases {
-        assert!(lease.layout.end_offset <= arena_config().reservation_ceiling);
-        assert_eq!(lease.layout.start_offset % lease.requested_alignment, 0);
-    }
-
-    // Re-observing the same live lease must not change offsets.
-    let before = leases[0].layout.clone();
-    let after = harness
-        .lookup_lease(leases[0].ring_id)
-        .expect("live lease must be lookupable")
-        .layout
-        .clone();
-    assert_eq!(after, before);
-}
-
-// This proves CancelLease removes queued work and suppresses later hot-path
-// installation, including a fresh lease that races with cancellation.
-#[test]
-fn cancelled_queued_lease_never_installs_hot_path_state() {
-    // Fill the arena and queue a second satisfiable lease.
-    let mut harness = new_arena();
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 3072, 64)));
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64)));
-
-    // Cancel the queued request before it is leased.
-    harness.request(arena::ArenaRequest::CancelLease {
-        request_id: arena::LeaseRequestId(2),
-    });
-
-    // Releasing pressure must not install worker or pump state for the canceled
-    // request.
-    let live = harness.live_leases()[0].ring_id;
-    harness.request(arena::ArenaRequest::ReleaseRing {
-        ring_id: live,
-        proof: arena::QuiescenceProof::verified(),
-    });
-    assert!(!harness.commands().iter().any(|command| {
-        matches!(
-            command,
-            arena::ArenaCommand::InstallWorkerOrPumpState {
-                request_id: arena::LeaseRequestId(2),
-                ..
-            }
-        )
-    }));
-
-    // If a fresh lease was produced during the race, it must be released instead
-    // of becoming hot-path state.
-    assert!(harness.events().iter().any(|event| {
-        matches!(
-            event,
-            arena::ArenaEvent::CancelledFreshLeaseReleased {
-                request_id: arena::LeaseRequestId(2)
-            }
-        )
-    }));
-}
-
-// This proves ranges are released only with quiescence proof, are not reused
-// while live work owns them, and may be reused after release.
-#[test]
-fn release_requires_quiescence_and_reuse_happens_only_after_release() {
-    // Lease one ring and record its range.
-    let mut harness = new_arena();
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 1024, 64)));
-    let first = harness.live_leases()[0].clone();
-
-    // Release without proof must reject and keep the range live.
-    harness.request(arena::ArenaRequest::ReleaseRing {
-        ring_id: first.ring_id,
-        proof: arena::QuiescenceProof::missing(),
-    });
-    assert!(harness.lookup_lease(first.ring_id).is_some());
-
-    // A second lease while the first is live must not overlap the first range.
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64)));
-    let second = harness
-        .live_leases()
-        .iter()
-        .find(|lease| lease.ring_id != first.ring_id)
-        .expect("second live lease must exist")
-        .clone();
-    assert_non_overlapping(&first.layout, &second.layout);
-
-    // Verified quiescence allows release, after which reuse is legal.
-    harness.request(arena::ArenaRequest::ReleaseRing {
-        ring_id: first.ring_id,
-        proof: arena::QuiescenceProof::verified(),
-    });
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(3, 1024, 64)));
-    assert!(
-        harness
-            .events()
-            .iter()
-            .any(|event| { matches!(event, arena::ArenaEvent::RingLeased { .. }) })
-    );
-}
-
-// This proves shutdown rejects new leases, preserves live lease records, and
-// does not release live ranges without quiescence proof.
-#[test]
-fn shutdown_rejects_new_leases_without_corrupting_live_records() {
-    // Create a live lease before shutdown.
-    let mut harness = new_arena();
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
-    let live_before = harness.live_leases().to_vec();
-
-    // Shut the arena down.
-    harness.request(arena::ArenaRequest::Shutdown);
-
-    // New leases are rejected after shutdown.
-    harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 512, 64)));
-    assert!(harness.events().iter().any(|event| {
-        matches!(
-            event,
-            arena::ArenaEvent::RingLeaseRejected {
-                request_id: arena::LeaseRequestId(2),
-                reason: arena::RingLeaseRejection::ArenaShuttingDown,
-            }
-        )
-    }));
-
-    // Existing live lease records remain intact until proof-backed release.
-    assert_eq!(harness.live_leases(), live_before.as_slice());
-}

crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs

@@ -0,0 +1,96 @@
+use data_plane::actor as dp;
+use mvp_system::node::actor::NodeAgentMsg;
+use mvp_system::node::data_plane_bridge::MvpDataPlaneReportSinkActor;
+use swactor::config::RuntimeConfig;
+use swactor::runtime::Runtime;
+
+fn spawn_bridge(
+    runtime: &Runtime,
+) -> (
+    swactor::actor::ActorAddress,
+    swactor::runtime::Inbox<NodeAgentMsg>,
+) {
+    let node_messages = runtime
+        .new_inbox::<NodeAgentMsg>()
+        .expect("node-agent inbox");
+    let bridge = runtime
+        .spawn(MvpDataPlaneReportSinkActor::new(*node_messages.addr()))
+        .expect("spawn bridge");
+    (bridge, node_messages)
+}
+
+#[test]
+fn data_plane_reports_drive_mvp_stage_readiness_and_object_visibility() {
+    let runtime = Runtime::new(RuntimeConfig::default());
+    let (bridge, node_messages) = spawn_bridge(&runtime);
+
+    runtime
+        .send_to(
+            bridge,
+            dp::DataPlaneReportMsg::InboundEdgeReady {
+                edge_id: dp::EdgeId(7001),
+            },
+        )
+        .expect("send inbound ready");
+    runtime
+        .send_to(
+            bridge,
+            dp::DataPlaneReportMsg::ObjectLoaded {
+                edge_id: dp::EdgeId(7001),
+                object_id: data_plane::object_record::ObjectId(9000),
+                sequence: 3,
+                extent: 8,
+                handle: dp::DeviceHandle::new(dp::WorkerGeneration(2), 44),
+            },
+        )
+        .expect("send object loaded");
+    runtime.tick();
+
+    assert_eq!(
+        node_messages.try_recv(),
+        Some(NodeAgentMsg::MarkInboundEdgeReady { edge_id: 7001 })
+    );
+    assert_eq!(
+        node_messages.try_recv(),
+        Some(NodeAgentMsg::ObjectLoaded {
+            edge_id: 7001,
+            object_id: 9000,
+            sequence: 3,
+            handle_generation: 2,
+            handle_id: 44,
+        })
+    );
+}
+
+#[test]
+fn data_plane_fault_and_stop_reports_map_to_mvp_lifecycle_messages() {
+    let runtime = Runtime::new(RuntimeConfig::default());
+    let (bridge, node_messages) = spawn_bridge(&runtime);
+
+    runtime
+        .send_to(
+            bridge,
+            dp::DataPlaneReportMsg::EdgeFaulted {
+                edge_id: dp::EdgeId(7001),
+                reason: dp::EdgeFaultReason::StreamFault(
+                    data_plane::edge_lifecycle::StreamFaultReason::ProtocolError,
+                ),
+            },
+        )
+        .expect("send edge fault");
+    runtime
+        .send_to(
+            bridge,
+            dp::DataPlaneReportMsg::LocalEdgesStopped {
+                run_id: dp::RunId(55),
+            },
+        )
+        .expect("send stopped");
+    runtime.tick();
+
+    assert_eq!(node_messages.try_recv(), Some(NodeAgentMsg::WorkerCrashed));
+    assert_eq!(
+        node_messages.try_recv(),
+        Some(NodeAgentMsg::LocalEdgesStopped { run_id: 55 })
+    );
+}

crates/mvp-system/src/tests/gpu_worker_ingress_parser_guarantees.rs

@@ -9,7 +9,7 @@
 //! They assert the guarantees in
 //! `specs/mvp_system/gpu_worker_ingress_parser_contract.md`.
 
-use mvp_system::node_data::ingress;
+use data_plane::ingress;
 
 // A valid ingress ring config supplies the edge object spec and current worker
 // generation. The parser remains a black box behind ring helper operations.

crates/mvp-system/src/tests/local_mock/environment.rs

@@ -1,6 +1,6 @@
 use std::collections::{BTreeMap, BTreeSet};
 
-use mvp_system::node_data::edge_actor;
+use data_plane::edge_actor;
 use mvp_system::observability::lifecycle as obs;
 use mvp_system::orchestration::engine_builder as engine;
 use mvp_system::orchestration::run_fsm as fsm;

crates/mvp-system/src/tests/local_mock/mock_node.rs

@@ -1,4 +1,4 @@
-use mvp_system::node_data::edge_actor;
+use data_plane::edge_actor;
 use mvp_system::orchestration::run_plan as plan;
 use mvp_system::staging as stage;
 

crates/mvp-system/src/tests/mod.rs

@@ -1,11 +1,9 @@
-mod arena_manager_guarantees;
 mod bootstrap_datastream_guarantees;
+mod data_plane_bridge_guarantees;
 mod device_bridge_guarantees;
 mod docker_cluster_provisioning_guarantees;
-mod edge_establisher_guarantees;
 mod engine_builder_guarantees;
 mod gpu_worker_ctl_guarantees;
-mod gpu_worker_egress_producer_guarantees;
 mod gpu_worker_ingress_parser_guarantees;
 mod gpu_worker_process_adapter_guarantees;
 mod local_mock;

crates/mvp-system/src/tests/module_surface_guarantees.rs

@@ -3,9 +3,8 @@
 //! These tests assert only the observable public-path contract: target modules
 //! expose existing behavior without introducing copied type definitions.
 
-use mvp_system::{
+use data_plane::{arena as data_arena, edge_lifecycle as data_edge};
-    chat, node, node_data, observability, orchestration, prompt, staging, transport, worker,
+use mvp_system::{chat, node, observability, orchestration, prompt, staging, transport, worker};
-};
 
 #[test]
 fn target_modules_offer_new_paths_to_existing_public_contracts() {
@@ -15,14 +14,14 @@ fn target_modules_offer_new_paths_to_existing_public_contracts() {
     let boot_node_id: node::boot_lifecycle::NodeId = node::boot_lifecycle::NodeId(11);
     assert_eq!(boot_node_id, node::boot_lifecycle::NodeId(11));
 
-    let arena_ring_id: node_data::arena::RingId = node_data::arena::RingId(3);
+    let arena_ring_id: data_arena::RingId = data_arena::RingId(3);
-    assert_eq!(arena_ring_id, node_data::arena::RingId(3));
+    assert_eq!(arena_ring_id, data_arena::RingId(3));
 
     let stage_edge_id: staging::EdgeId = staging::EdgeId(7001);
     assert_eq!(stage_edge_id, staging::EdgeId(7001));
 
-    let transport_edge_id: node::edge_lifecycle::EdgeId = node::edge_lifecycle::EdgeId(7002);
+    let transport_edge_id: data_edge::EdgeId = data_edge::EdgeId(7002);
-    assert_eq!(transport_edge_id, node::edge_lifecycle::EdgeId(7002));
+    assert_eq!(transport_edge_id, data_edge::EdgeId(7002));
 
     let worker_generation: worker::WorkerGeneration = worker::WorkerGeneration(2);
     assert_eq!(worker_generation, worker::WorkerGeneration(2));

crates/mvp-system/src/tests/shared_ring_helper_abi_guarantees.rs

@@ -9,7 +9,7 @@
 //! They assert the guarantees in
 //! `specs/mvp_system/shared_ring_helper_abi_contract.md`.
 
-use mvp_system::node_data::ring;
+use data_plane::ring;
 
 // A small ring forces wraparound and full/empty transitions quickly. The helper
 // still owns the actual shared-memory atomics and process-local address math.

crates/mvp-system/src/tests/tx_rx_edge_actor_guarantees.rs

@@ -8,7 +8,7 @@
 //! They assert the guarantees in
 //! `specs/mvp_system/tx_rx_edge_actor_contract.md`.
 
-use mvp_system::node_data::edge_actor;
+use data_plane::edge_actor;
 
 // The edge id fixture gives both actors a shared identity while keeping Tx and
 // Rx lifecycle tests independent from driver and ring internals.

crates/mvp-system/src/tests/worker_edge_adapter_guarantees.rs

@@ -1,9 +1,9 @@
+use data_plane::object_record as ingress;
 use mvp_system::node::actor::{
     NodeAgentActor, NodeAgentMsg, NodeAgentReport, StageCommandWire, StageEdgeKindWire,
     StageInboundEdgeWire, StageObjectSpecWire, StageOutboundEdgeWire, StageProvisionWire,
     StageRingSpecWire,
 };
-use mvp_system::node_data::object as ingress;
 use mvp_system::orchestration::run_plan::{self, GgufSource, TokenizerSource};
 use mvp_system::staging as stage;
 use swactor::actor::ActorAddress;

crates/mvp-system/src/worker/mod.rs

@@ -2,7 +2,6 @@
 
 pub mod control;
 pub mod device_bridge;
-pub mod egress;
 pub mod process_adapter;
 
 pub mod process {