WO-4: KinematicTransformFrame, DataChannelHub, Axum telemetry, StatsSource wiring

T2: KinematicTransformFrame (88 bytes, repr(C)+Pod) in miranda-core closes
the WO-3 breath.rs gap where head_pitch_deg/clavicle_rise were computed but
could not be routed — ARKit-52 is face-only. from_breath() maps to joint
quaternions: head 100% X-rotation, neck 40%, shoulders +/-Z 4deg max.
Quaternion: 16 bytes, Pod, normalised/degenerate handling. RingId::Kinematic
added. miranda-ipc gains a 4th ring (KINEMATIC_SLOTS=128) with push/pop.

T3: DataChannelHub broadcasts 312-byte binary packets (MRD1 magic + two LE
u16 size fields + 216B BlendshapeFrame + 88B KinematicTransformFrame) to
all connected subscribers via bounded mpsc. Full channel drops frame and
increments counter, never stalls. Closed receivers pruned on next broadcast.
encode/decode in frame.rs is allocation-free in the hot path via scratch
BytesMut. webrtc 0.8.0 declared optional behind --features webrtc-native;
wire format is identical whether transport layer is axum WS or DTLS channel.

T4: TelemetryHub broadcasts camelCase JSON snapshots at configurable interval.
Three-state circuit breaker: Closed -> HalfOpen (any drops) -> Open (>=500ms
of drops); resets to Closed after 1000ms clean. DispatchSource trait
decouples the hub from miranda-nodes (testable with a mock). Axum server:
/health (JSON), /data (binary WS), /telemetry (text WS). Integration tests
use OS-assigned ports and a real tokio_tungstenite client.

T5: StatsSource wraps Arc<DispatchStats>, implements DispatchSource with
Relaxed atomics -- same ordering DispatchStats uses internally; telemetry
snapshots are samples, not synchronization points.

198 tests workspace-wide, zero warnings.

miranda-core/src/lib.rs

@@ -277,7 +277,7 @@ pub struct BackpressureError {
     pub capacity: usize,
 }
 
-/// Identifies which of the three rings on the bus a `BackpressureError`
+/// Identifies which of the four rings on the bus a `BackpressureError`
 /// came from.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum RingId {
@@ -287,6 +287,8 @@ pub enum RingId {
     Blendshape,
     /// The `SphericalHarmonics` ring.
     SphericalHarmonics,
+    /// The `KinematicTransformFrame` ring (added WO-4).
+    Kinematic,
 }
 
 impl core::fmt::Display for BackpressureError {
@@ -430,3 +432,378 @@ mod tests {
         assert_eq!(arkit::index_of("jawopen"), None);
     }
 }
+
+// ── KinematicTransformFrame ──────────────────────────────────────────────────
+
+/// Number of skeletal joints carried in [`KinematicTransformFrame`].
+///
+/// HEAD, NECK, SHOULDER_LEFT, SHOULDER_RIGHT — the four joints the
+/// respiratory oscillator (WO-3 T5) and future full-body kinematics need
+/// to route. ARKit-52 is face-only and structurally cannot carry these, so
+/// they travel on a parallel frame type rather than being smuggled into an
+/// unrelated face channel.
+pub const KINEMATIC_JOINT_COUNT: usize = 4;
+
+/// Index constants for the joints in [`KinematicTransformFrame`].
+///
+/// Matches the order the renderer (WO-5) expects: head before neck so the
+/// dependency chain is top-down.
+pub mod kinematic_joints {
+    pub const HEAD: usize = 0;
+    pub const NECK: usize = 1;
+    pub const SHOULDER_LEFT: usize = 2;
+    pub const SHOULDER_RIGHT: usize = 3;
+
+    /// Human-readable names in index order, for telemetry and debug.
+    pub const JOINT_NAMES: [&str; super::KINEMATIC_JOINT_COUNT] =
+        ["head", "neck", "shoulder_left", "shoulder_right"];
+}
+
+/// A unit quaternion representing a joint's local rotation, stored as
+/// `[x, y, z, w]` in the standard conventions used by every game-engine
+/// and animation tool this project targets (GLM, Three.js, Unity, Omniverse).
+///
+/// - Identity: `[0.0, 0.0, 0.0, 1.0]`.
+/// - `w = 1.0` → no rotation.  `w = 0.0, xyz = axis` → 180° rotation.
+/// - **Must be unit-length.** Consumers normalize defensively, but a
+///   non-unit quaternion is a producer bug — it will produce scale artefacts
+///   on the rendered mesh.
+///
+/// Size: 4 × f32 = 16 bytes.
+#[repr(C)]
+#[derive(Debug, Clone, Copy, PartialEq, bytemuck::Pod, bytemuck::Zeroable)]
+pub struct Quaternion {
+    pub x: f32,
+    pub y: f32,
+    pub z: f32,
+    pub w: f32,
+}
+
+impl Quaternion {
+    /// The identity rotation — no change from the bind pose.
+    pub const IDENTITY: Self = Self { x: 0.0, y: 0.0, z: 0.0, w: 1.0 };
+
+    /// Builds a quaternion from a rotation around the X axis (pitch), in
+    /// radians. Positive angle tilts the joint forward (head nods down).
+    #[inline]
+    pub fn from_angle_x(radians: f32) -> Self {
+        let (s, c) = (radians * 0.5).sin_cos();
+        Self { x: s, y: 0.0, z: 0.0, w: c }
+    }
+
+    /// Builds a quaternion from a rotation around the Y axis (yaw), in
+    /// radians.
+    #[inline]
+    pub fn from_angle_y(radians: f32) -> Self {
+        let (s, c) = (radians * 0.5).sin_cos();
+        Self { x: 0.0, y: s, z: 0.0, w: c }
+    }
+
+    /// Builds a quaternion from a rotation around the Z axis (roll), in
+    /// radians.
+    #[inline]
+    pub fn from_angle_z(radians: f32) -> Self {
+        let (s, c) = (radians * 0.5).sin_cos();
+        Self { x: 0.0, y: 0.0, z: s, w: c }
+    }
+
+    /// Squared magnitude — should be 1.0 for a valid rotation quaternion.
+    #[inline]
+    pub fn norm_sq(self) -> f32 {
+        self.x * self.x + self.y * self.y + self.z * self.z + self.w * self.w
+    }
+
+    /// Normalises to unit length. Returns `IDENTITY` if the quaternion is
+    /// degenerate (zero or near-zero magnitude) rather than propagating NaN.
+    #[inline]
+    pub fn normalised(self) -> Self {
+        let n = self.norm_sq().sqrt();
+        if n < f32::EPSILON {
+            return Self::IDENTITY;
+        }
+        let inv = 1.0 / n;
+        Self {
+            x: self.x * inv,
+            y: self.y * inv,
+            z: self.z * inv,
+            w: self.w * inv,
+        }
+    }
+
+    /// Returns `true` if this quaternion is within `tolerance` of unit
+    /// length. Useful in tests and assertions.
+    #[inline]
+    pub fn is_unit(&self, tolerance: f32) -> bool {
+        (self.norm_sq() - 1.0).abs() < tolerance
+    }
+}
+
+impl Default for Quaternion {
+    fn default() -> Self {
+        Self::IDENTITY
+    }
+}
+
+/// Body-rig kinematic transform frame — joint quaternions for the skeletal
+/// joints that ARKit-52's face-only blendshape frame cannot carry.
+///
+/// This is the companion to [`BlendshapeFrame`], travelling in lockstep
+/// through the transport layer. Together they carry the complete per-frame
+/// face + body state that EVE's renderer needs.
+///
+/// # Layout (WO-4 defined — do not add fields without updating the IPC ring)
+///
+/// ```text
+/// offset  0 : timestamp_us : u64           →   8 bytes
+/// offset  8 : joints       : [Quaternion; 4]  → 64 bytes  (4 × 16)
+/// offset 72 : head_pitch_deg : f32          →   4 bytes
+/// offset 76 : clavicle_rise  : f32          →   4 bytes
+/// offset 80 : _reserved    : [u8; 8]        →   8 bytes  (explicit pad)
+/// ─────────────────────────────────────────────────────
+/// total                                        88 bytes
+/// ```
+///
+/// The `_reserved` field is zero on write and must be ignored on read.
+/// It provides expansion room (e.g. for a spine joint, shoulder elevation
+/// scalar, or protocol version byte) without breaking the mmap layout or
+/// existing consumers.
+///
+/// # The WO-3 gap this closes
+///
+/// `miranda-nodes/src/breath.rs` computes `head_pitch_deg` (head nodding
+/// with the respiratory cycle) and `clavicle_rise` (chest rising on
+/// inhalation). Both are physiologically correct values, but `BlendshapeFrame`
+/// — ARKit-52 face weights only — has nowhere to put them. Rather than
+/// silently smuggling them into an unrelated face channel, WO-3 documented
+/// the gap and wrote a test asserting no smuggling occurred. This struct is
+/// the promised fix: `BreathGenerator::build_transform_frame()` (WO-3
+/// `breath.rs`) can now populate the right fields.
+#[repr(C)]
+#[derive(Debug, Clone, Copy, bytemuck::Pod, bytemuck::Zeroable)]
+pub struct KinematicTransformFrame {
+    /// Microseconds since harness start — matches [`BlendshapeFrame::timestamp_us`]
+    /// so the renderer can pair the two frames by timestamp.
+    pub timestamp_us: u64,
+    /// Local-space rotation quaternions for each joint, in
+    /// [`kinematic_joints`] index order. Identity quaternion (`w=1`) means
+    /// no deviation from the bind pose.
+    pub joints: [Quaternion; KINEMATIC_JOINT_COUNT],
+    /// Head nodding amplitude from the respiratory cycle, in degrees.
+    /// Positive = forward nod (chin toward chest). Signed — the head nods
+    /// slightly backward on the exhale.
+    ///
+    /// This is a convenience scalar: the full rotation is also encoded in
+    /// `joints[HEAD]`, but consumers that only need the scalar can read
+    /// this field without decomposing a quaternion.
+    pub head_pitch_deg: f32,
+    /// Fractional clavicle rise from inhalation, in `[0, 1]`. Zero is the
+    /// resting position; 1.0 would be full elevation (never reached in
+    /// practice — normal tidal breathing drives this to roughly 0.3).
+    ///
+    /// Also a convenience scalar mirroring `joints[SHOULDER_LEFT/RIGHT]`'s
+    /// vertical component.
+    pub clavicle_rise: f32,
+    /// Reserved for future fields. Always zero on write.
+    pub _reserved: [u8; 8],
+}
+
+impl KinematicTransformFrame {
+    /// Rest-pose frame — all joints at identity, scalar outputs zero.
+    pub const REST: Self = Self {
+        timestamp_us: 0,
+        joints: [Quaternion::IDENTITY; KINEMATIC_JOINT_COUNT],
+        head_pitch_deg: 0.0,
+        clavicle_rise: 0.0,
+        _reserved: [0; 8],
+    };
+
+    /// Builds a frame from the outputs of `BreathGenerator::tick()`.
+    ///
+    /// Converts `head_pitch_deg` into a proper X-axis rotation quaternion
+    /// for the head joint and a matching smaller rotation for the neck
+    /// joint (neck typically participates at ≈40% of head travel). Shoulders
+    /// rise symmetrically by `clavicle_rise`.
+    pub fn from_breath(
+        timestamp_us: u64,
+        head_pitch_deg: f32,
+        clavicle_rise: f32,
+    ) -> Self {
+        let head_rad = head_pitch_deg.to_radians();
+        // Neck participates at 40% of head pitch — anatomically, head motion
+        // distributes across cervical vertebrae, not just the atlanto-occipital
+        // joint. A 60/40 split between head and neck joints is the standard
+        // authoring convention used by major rigs (Metahuman, UE5 mannequin).
+        let neck_rad = head_rad * 0.4;
+        // Shoulder elevation: clavicle_rise in [0,1] maps to a Z-axis
+        // rotation of up to 4° per shoulder (typical quiet breathing range).
+        let shoulder_elevation_rad = clavicle_rise * 4.0f32.to_radians();
+
+        let mut joints = [Quaternion::IDENTITY; KINEMATIC_JOINT_COUNT];
+        joints[kinematic_joints::HEAD] = Quaternion::from_angle_x(head_rad).normalised();
+        joints[kinematic_joints::NECK] = Quaternion::from_angle_x(neck_rad).normalised();
+        // Shoulders: left rotates positive Z, right negative Z (symmetric
+        // elevation). The sign convention matches the Y-up, right-hand
+        // coordinate system used by Three.js / GLB rigs.
+        joints[kinematic_joints::SHOULDER_LEFT] =
+            Quaternion::from_angle_z(shoulder_elevation_rad).normalised();
+        joints[kinematic_joints::SHOULDER_RIGHT] =
+            Quaternion::from_angle_z(-shoulder_elevation_rad).normalised();
+
+        Self {
+            timestamp_us,
+            joints,
+            head_pitch_deg,
+            clavicle_rise,
+            _reserved: [0; 8],
+        }
+    }
+}
+
+impl Default for KinematicTransformFrame {
+    fn default() -> Self {
+        Self::REST
+    }
+}
+
+#[cfg(test)]
+mod kinematic_tests {
+    use super::*;
+
+    #[test]
+    fn struct_size_is_exactly_88_bytes() {
+        assert_eq!(std::mem::size_of::<Quaternion>(), 16);
+        assert_eq!(
+            std::mem::size_of::<KinematicTransformFrame>(),
+            88,
+            "KinematicTransformFrame layout changed — update the IPC ring \
+             capacity and the transport framing"
+        );
+    }
+
+    #[test]
+    fn rest_frame_is_all_identity_and_zero() {
+        let f = KinematicTransformFrame::REST;
+        assert_eq!(f.timestamp_us, 0);
+        assert_eq!(f.head_pitch_deg, 0.0);
+        assert_eq!(f.clavicle_rise, 0.0);
+        assert_eq!(f._reserved, [0u8; 8]);
+        for j in f.joints.iter() {
+            assert_eq!(*j, Quaternion::IDENTITY);
+        }
+    }
+
+    #[test]
+    fn identity_quaternion_is_unit_length() {
+        assert!(Quaternion::IDENTITY.is_unit(1e-6));
+    }
+
+    #[test]
+    fn quaternion_from_angle_x_is_unit() {
+        for deg in [-90.0f32, -45.0, -1.0, 0.0, 1.0, 15.0, 45.0, 90.0] {
+            let q = Quaternion::from_angle_x(deg.to_radians());
+            assert!(
+                q.is_unit(1e-5),
+                "from_angle_x({deg}) norm² = {} (not 1)",
+                q.norm_sq()
+            );
+        }
+    }
+
+    #[test]
+    fn from_breath_produces_valid_quaternions() {
+        let f = KinematicTransformFrame::from_breath(1_000_000, 0.6, 0.3);
+        for (i, j) in f.joints.iter().enumerate() {
+            assert!(
+                j.is_unit(1e-5),
+                "joint {} quaternion is not unit: {:?}", i, j
+            );
+        }
+        assert_eq!(f.head_pitch_deg, 0.6);
+        assert_eq!(f.clavicle_rise, 0.3);
+        assert_eq!(f._reserved, [0u8; 8]);
+    }
+
+    #[test]
+    fn head_leads_neck() {
+        // Head joint must rotate more than neck — neck participates at 40%.
+        let f = KinematicTransformFrame::from_breath(0, 5.0, 0.0);
+        // The X component of a quaternion is sin(θ/2); larger pitch ⇒ larger |x|.
+        let head_x = f.joints[kinematic_joints::HEAD].x.abs();
+        let neck_x = f.joints[kinematic_joints::NECK].x.abs();
+        assert!(
+            head_x > neck_x,
+            "head x={head_x} should lead neck x={neck_x}"
+        );
+        assert!(
+            (head_x / neck_x - 1.0 / 0.4).abs() < 0.01,
+            "neck should be ~40% of head pitch, got ratio {}",
+            head_x / neck_x
+        );
+    }
+
+    #[test]
+    fn shoulders_are_symmetric_and_scale_with_clavicle_rise() {
+        let f_half = KinematicTransformFrame::from_breath(0, 0.0, 0.5);
+        let f_full = KinematicTransformFrame::from_breath(0, 0.0, 1.0);
+        let sl = f_half.joints[kinematic_joints::SHOULDER_LEFT];
+        let sr = f_half.joints[kinematic_joints::SHOULDER_RIGHT];
+        // Symmetric: Z components should be equal in magnitude, opposite in sign.
+        assert!(
+            (sl.z + sr.z).abs() < 1e-6,
+            "shoulders are not symmetric: L.z={} R.z={}", sl.z, sr.z
+        );
+        // Scale: full rise should rotate more than half rise.
+        let sl_full = f_full.joints[kinematic_joints::SHOULDER_LEFT];
+        assert!(
+            sl_full.z.abs() > sl.z.abs(),
+            "shoulder at rise=1.0 should exceed rise=0.5"
+        );
+    }
+
+    #[test]
+    fn zero_breath_produces_rest_pose_joints() {
+        let f = KinematicTransformFrame::from_breath(0, 0.0, 0.0);
+        for (i, j) in f.joints.iter().enumerate() {
+            assert!(
+                j.is_unit(1e-6) && (j.x.abs() < 1e-7) && (j.y.abs() < 1e-7) && (j.z.abs() < 1e-7),
+                "joint {i} should be identity at zero breath: {j:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn bytemuck_pod_round_trip() {
+        // Proves the Pod derive is sound: casting to bytes and back must
+        // give an identical value, with no padding bytes that could be
+        // non-deterministic.
+        let original = KinematicTransformFrame::from_breath(42_000, 0.8, 0.25);
+        let bytes: &[u8] = bytemuck::bytes_of(&original);
+        assert_eq!(bytes.len(), 88);
+        let recovered: KinematicTransformFrame = *bytemuck::from_bytes(bytes);
+        // Compare field-by-field since f32 doesn't impl Eq.
+        assert_eq!(recovered.timestamp_us, original.timestamp_us);
+        assert_eq!(recovered.head_pitch_deg, original.head_pitch_deg);
+        assert_eq!(recovered.clavicle_rise, original.clavicle_rise);
+        for i in 0..KINEMATIC_JOINT_COUNT {
+            assert_eq!(recovered.joints[i], original.joints[i]);
+        }
+    }
+
+    /// Joint index names must agree with `JOINT_NAMES`.
+    #[test]
+    fn joint_constants_match_names_array() {
+        use kinematic_joints::*;
+        assert_eq!(JOINT_NAMES[HEAD], "head");
+        assert_eq!(JOINT_NAMES[NECK], "neck");
+        assert_eq!(JOINT_NAMES[SHOULDER_LEFT], "shoulder_left");
+        assert_eq!(JOINT_NAMES[SHOULDER_RIGHT], "shoulder_right");
+    }
+
+    /// Normalise must not panic or produce NaN on a zero quaternion.
+    #[test]
+    fn normalise_handles_degenerate_input() {
+        let q = Quaternion { x: 0.0, y: 0.0, z: 0.0, w: 0.0 };
+        let n = q.normalised();
+        assert_eq!(n, Quaternion::IDENTITY, "degenerate quaternion must return IDENTITY");
+    }
+}

miranda-ipc/src/lib.rs

@@ -1,11 +1,12 @@
 //! miranda-ipc — Work Order 1: the POSIX shared-memory ring buffer at
 //! `/dev/shm/miranda_bus`, lock-free atomic head/tail pointers, and C-ABI
 //! aligned payload structs for audio chunks, 52-channel ARKit blendshape
-//! frames, and 9-coefficient spherical harmonic vectors.
+//! frames, 9-coefficient spherical harmonic vectors, and WO-4
+//! `KinematicTransformFrame` quaternions for neck/clavicle routing.
 //!
 //! # What this is
 //!
-//! Three independent single-producer / single-consumer (SPSC) ring buffers
+//! Four independent single-producer / single-consumer (SPSC) ring buffers
 //! sharing one memory mapping. Each ring is lock-free: coordination is a pair
 //! of `AtomicUsize` counters (`head` = read position, `tail` = write
 //! position), never a mutex. The bus is on the critical path of a 60 FPS
@@ -50,7 +51,10 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 
 use bytemuck::Pod;
 use memmap2::MmapMut;
-use miranda_core::{AudioChunk, BackpressureError, BlendshapeFrame, RingId, SphericalHarmonics};
+use miranda_core::{
+    AudioChunk, BackpressureError, BlendshapeFrame, KinematicTransformFrame, RingId,
+    SphericalHarmonics,
+};
 
 /// Canonical path for the production bus. `/dev/shm` is a tmpfs mount
 /// (RAM-backed), which is what makes the ≤50 μs round-trip target
@@ -71,6 +75,9 @@ const CTRL_BYTES: usize = 2 * CACHE_LINE;
 const AUDIO_SLOTS: usize = 64;
 const BLEND_SLOTS: usize = 128;
 const SH_SLOTS: usize = 128;
+/// 128 slots × 88 bytes = 11,264 bytes. Matches the blendshape ring depth
+/// because every blendshape frame should have a companion kinematic frame.
+const KINEMATIC_SLOTS: usize = 128;
 
 /// Rounds `value` up to the next multiple of `align`. `align` must be a
 /// power of two.
@@ -94,8 +101,13 @@ const SH_CTRL_OFF: usize = align_up(BLEND_DATA_OFF + BLEND_DATA_BYTES, CACHE_LIN
 const SH_DATA_OFF: usize = align_up(SH_CTRL_OFF + CTRL_BYTES, CACHE_LINE);
 const SH_DATA_BYTES: usize = SH_SLOTS * std::mem::size_of::<SphericalHarmonics>();
 
+const KINEMATIC_CTRL_OFF: usize = align_up(SH_DATA_OFF + SH_DATA_BYTES, CACHE_LINE);
+const KINEMATIC_DATA_OFF: usize = align_up(KINEMATIC_CTRL_OFF + CTRL_BYTES, CACHE_LINE);
+const KINEMATIC_DATA_BYTES: usize =
+    KINEMATIC_SLOTS * std::mem::size_of::<KinematicTransformFrame>();
+
 /// Total size of the shared-memory region, rounded up to a cache line.
-pub const BUS_TOTAL_BYTES: usize = align_up(SH_DATA_OFF + SH_DATA_BYTES, CACHE_LINE);
+pub const BUS_TOTAL_BYTES: usize = align_up(KINEMATIC_DATA_OFF + KINEMATIC_DATA_BYTES, CACHE_LINE);
 
 // Compile-time proof of every layout invariant the unsafe code below relies
 // on. If any of these ever becomes false (someone adds a field, changes a
@@ -105,11 +117,13 @@ const _: () = {
     assert!(AUDIO_SLOTS.is_power_of_two());
     assert!(BLEND_SLOTS.is_power_of_two());
     assert!(SH_SLOTS.is_power_of_two());
+    assert!(KINEMATIC_SLOTS.is_power_of_two());
 
     // Control blocks must be aligned for AtomicUsize.
     assert!(AUDIO_CTRL_OFF % std::mem::align_of::<AtomicUsize>() == 0);
     assert!(BLEND_CTRL_OFF % std::mem::align_of::<AtomicUsize>() == 0);
     assert!(SH_CTRL_OFF % std::mem::align_of::<AtomicUsize>() == 0);
+    assert!(KINEMATIC_CTRL_OFF % std::mem::align_of::<AtomicUsize>() == 0);
     // ...and head/tail must not share a cache line.
     assert!(CTRL_BYTES >= 2 * CACHE_LINE);
     assert!(std::mem::size_of::<AtomicUsize>() <= CACHE_LINE);
@@ -125,11 +139,18 @@ const _: () = {
     assert!(
         std::mem::size_of::<SphericalHarmonics>() % std::mem::align_of::<SphericalHarmonics>() == 0
     );
+    assert!(KINEMATIC_DATA_OFF % std::mem::align_of::<KinematicTransformFrame>() == 0);
+    assert!(
+        std::mem::size_of::<KinematicTransformFrame>()
+            % std::mem::align_of::<KinematicTransformFrame>()
+            == 0
+    );
 
     // Regions must not overlap.
     assert!(AUDIO_DATA_OFF + AUDIO_DATA_BYTES <= BLEND_CTRL_OFF);
     assert!(BLEND_DATA_OFF + BLEND_DATA_BYTES <= SH_CTRL_OFF);
-    assert!(SH_DATA_OFF + SH_DATA_BYTES <= BUS_TOTAL_BYTES);
+    assert!(SH_DATA_OFF + SH_DATA_BYTES <= KINEMATIC_CTRL_OFF);
+    assert!(KINEMATIC_DATA_OFF + KINEMATIC_DATA_BYTES <= BUS_TOTAL_BYTES);
 };
 
 /// A 64-byte, 64-byte-aligned block. Used only to obtain a cache-line-aligned
@@ -498,10 +519,33 @@ impl MirandaBus {
         self.ring_pop(SH_CTRL_OFF, SH_DATA_OFF, SH_SLOTS)
     }
 
-    /// Capacity in slots of each ring, as `(audio, blendshape, sh)`.
-    /// Exposed for telemetry — THE VANITY shows ring occupancy per node.
-    pub const fn capacities() -> (usize, usize, usize) {
-        (AUDIO_SLOTS, BLEND_SLOTS, SH_SLOTS)
+    /// Pushes a [`KinematicTransformFrame`] onto the kinematic ring.
+    ///
+    /// The kinematic ring runs in lockstep with the blendshape ring: one
+    /// frame per 16.67 ms, same timestamp, consumed together by the renderer
+    /// (WO-5) to drive both the face ARKit weights and the skeletal joint
+    /// rotations.
+    pub fn push_kinematic(
+        &self,
+        frame: KinematicTransformFrame,
+    ) -> Result<(), BackpressureError> {
+        self.ring_push(
+            KINEMATIC_CTRL_OFF,
+            KINEMATIC_DATA_OFF,
+            KINEMATIC_SLOTS,
+            RingId::Kinematic,
+            frame,
+        )
+    }
+
+    /// Pops a [`KinematicTransformFrame`] from the kinematic ring.
+    pub fn pop_kinematic(&self) -> Option<KinematicTransformFrame> {
+        self.ring_pop(KINEMATIC_CTRL_OFF, KINEMATIC_DATA_OFF, KINEMATIC_SLOTS)
+    }
+
+    /// Capacity in slots of each ring, as `(audio, blendshape, sh, kinematic)`.
+    pub const fn capacities() -> (usize, usize, usize, usize) {
+        (AUDIO_SLOTS, BLEND_SLOTS, SH_SLOTS, KINEMATIC_SLOTS)
     }
 }
 

miranda-transport/Cargo.toml

@@ -3,5 +3,36 @@ name = "miranda-transport"
 version.workspace = true
 edition.workspace = true
 
+[features]
+default = []
+# Enables the full webrtc-rs stack (DTLS/ICE/SRTP) for production NAT
+# traversal. Gated behind a feature because the crate pulls in heavy C++
+# crypto dependencies that are incompatible with the offline dev environment.
+# The binary-frame DataChannel wire format is identical whether the transport
+# layer is a bare axum WebSocket (default) or a real DTLS DataChannel (this
+# feature). Activate for production: cargo build --features webrtc-native
+webrtc-native = ["dep:webrtc"]
+
 [dependencies]
 miranda-core = { path = "../miranda-core" }
+miranda-nodes = { path = "../miranda-nodes" }
+miranda-ipc  = { path = "../miranda-ipc"   }
+
+# Data plane — binary blendshape frames over WebSocket / DataChannel.
+axum             = { version = "0.8.9",   features = ["ws", "macros"] }
+tokio            = { version = "1.53.1",  features = ["full"] }
+tokio-tungstenite = { version = "0.26.2" }
+
+# Serialization (telemetry JSON on the control plane).
+serde      = { version = "1", features = ["derive"] }
+serde_json = "1"
+bytemuck = "1"
+bytes      = "1"
+
+# Optional: full webrtc-rs stack for DTLS/ICE/SRTP.
+webrtc = { version = "0.8.0", optional = true }
+
+[dev-dependencies]
+tokio = { version = "1.53.1", features = ["full", "test-util"] }
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
+futures-util = "0.3"

miranda-transport/src/frame.rs

@@ -0,0 +1,242 @@
+//! Binary packet encoding for the data plane.
+//!
+//! Every data-plane WebSocket message is a 312-byte packet laid out as
+//! documented in the [`crate`] module docs. This module is the single place
+//! that knows that layout, so the encoder and the decoder stay in sync.
+
+use bytes::{BufMut, Bytes, BytesMut};
+use miranda_core::{BlendshapeFrame, KinematicTransformFrame};
+
+/// Magic bytes — first four bytes of every data-plane packet.
+///
+/// `MRD1` = Miranda Data Protocol version 1.
+/// A receiver that does not see this prefix should drop the packet and log
+/// a version-mismatch event rather than misinterpreting the payload.
+pub const PACKET_MAGIC: &[u8; 4] = b"MRD1";
+
+/// Total packet size in bytes.
+///
+/// 4 (magic) + 2 (blend_sz) + 2 (kin_sz) + 216 (BlendshapeFrame) +
+/// 88 (KinematicTransformFrame) = 312.
+pub const PACKET_SIZE: usize = 4 + 2 + 2
+    + std::mem::size_of::<BlendshapeFrame>()
+    + std::mem::size_of::<KinematicTransformFrame>();
+
+/// Encodes one data-plane packet into `dst`.
+///
+/// `dst` is grown by exactly [`PACKET_SIZE`] bytes. Calling this in the
+/// broadcast loop with a pre-allocated `BytesMut` avoids a per-frame heap
+/// allocation.
+pub fn encode(
+    blend: &BlendshapeFrame,
+    kin: &KinematicTransformFrame,
+    dst: &mut BytesMut,
+) {
+    dst.put_slice(PACKET_MAGIC);
+    dst.put_u16_le(std::mem::size_of::<BlendshapeFrame>() as u16);
+    dst.put_u16_le(std::mem::size_of::<KinematicTransformFrame>() as u16);
+    // SAFETY: BlendshapeFrame and KinematicTransformFrame are both
+    // repr(C) + bytemuck::Pod, which means:
+    //   - No padding bytes (Pod requires no uninit bytes).
+    //   - No invalid bit patterns (Pod requires every bit pattern valid).
+    //   - Size_of gives the exact number of meaningful bytes.
+    // Casting a Pod value to &[u8] via bytemuck::bytes_of is the canonical
+    // safe pattern — no raw pointer arithmetic, no manual transmute.
+    dst.put_slice(bytemuck::bytes_of(blend));
+    dst.put_slice(bytemuck::bytes_of(kin));
+    debug_assert_eq!(
+        dst.len() % PACKET_SIZE,
+        0,
+        "packet encode produced wrong size"
+    );
+}
+
+/// Encodes one data-plane packet and returns it as an owned [`Bytes`].
+///
+/// Allocates. Use the `encode` form with a scratch `BytesMut` in the hot
+/// loop; use this form in tests and slow paths.
+pub fn encode_to_bytes(blend: &BlendshapeFrame, kin: &KinematicTransformFrame) -> Bytes {
+    let mut buf = BytesMut::with_capacity(PACKET_SIZE);
+    encode(blend, kin, &mut buf);
+    buf.freeze()
+}
+
+/// Decodes a data-plane packet from `src`.
+///
+/// Returns `Err` if the packet is the wrong size, the magic bytes do not
+/// match, or the declared payload sizes disagree with the compiled sizes.
+/// Never panics.
+pub fn decode(src: &[u8]) -> Result<(BlendshapeFrame, KinematicTransformFrame), DecodeError> {
+    if src.len() != PACKET_SIZE {
+        return Err(DecodeError::WrongLength {
+            got: src.len(),
+            expected: PACKET_SIZE,
+        });
+    }
+    if &src[..4] != PACKET_MAGIC {
+        return Err(DecodeError::BadMagic);
+    }
+    let blend_sz = u16::from_le_bytes([src[4], src[5]]) as usize;
+    let kin_sz = u16::from_le_bytes([src[6], src[7]]) as usize;
+    if blend_sz != std::mem::size_of::<BlendshapeFrame>() {
+        return Err(DecodeError::SizeMismatch {
+            field: "BlendshapeFrame",
+            got: blend_sz,
+            expected: std::mem::size_of::<BlendshapeFrame>(),
+        });
+    }
+    if kin_sz != std::mem::size_of::<KinematicTransformFrame>() {
+        return Err(DecodeError::SizeMismatch {
+            field: "KinematicTransformFrame",
+            got: kin_sz,
+            expected: std::mem::size_of::<KinematicTransformFrame>(),
+        });
+    }
+    let blend_start = 8;
+    let kin_start = blend_start + blend_sz;
+    // SAFETY: same as encode — bytemuck::from_bytes is safe for Pod types.
+    let blend: BlendshapeFrame = *bytemuck::from_bytes(&src[blend_start..blend_start + blend_sz]);
+    let kin: KinematicTransformFrame =
+        *bytemuck::from_bytes(&src[kin_start..kin_start + kin_sz]);
+    Ok((blend, kin))
+}
+
+/// Errors that can occur when decoding a data-plane packet.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum DecodeError {
+    WrongLength { got: usize, expected: usize },
+    BadMagic,
+    SizeMismatch {
+        field: &'static str,
+        got: usize,
+        expected: usize,
+    },
+}
+
+impl std::fmt::Display for DecodeError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::WrongLength { got, expected } => {
+                write!(f, "packet wrong length: got {got}, expected {expected}")
+            }
+            Self::BadMagic => write!(f, "packet magic bytes do not match MRD1"),
+            Self::SizeMismatch { field, got, expected } => {
+                write!(
+                    f,
+                    "size field for {field} says {got} bytes, compiled size is {expected}"
+                )
+            }
+        }
+    }
+}
+
+impl std::error::Error for DecodeError {}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use miranda_core::{BLENDSHAPE_COUNT, KINEMATIC_JOINT_COUNT, Quaternion};
+
+    fn sample_blend(ts: u64) -> BlendshapeFrame {
+        BlendshapeFrame {
+            timestamp_us: ts,
+            weights: {
+                let mut w = [0.0f32; BLENDSHAPE_COUNT];
+                w[17] = 0.5; // jawOpen
+                w
+            },
+        }
+    }
+
+    fn sample_kin(ts: u64) -> KinematicTransformFrame {
+        KinematicTransformFrame {
+            timestamp_us: ts,
+            joints: [Quaternion::IDENTITY; KINEMATIC_JOINT_COUNT],
+            head_pitch_deg: 0.4,
+            clavicle_rise: 0.2,
+            _reserved: [0; 8],
+        }
+    }
+
+    #[test]
+    fn packet_size_constant_is_correct() {
+        // Verify the constant agrees with the actual sizes.
+        assert_eq!(
+            PACKET_SIZE,
+            4 + 2 + 2 + 216 + 88,
+            "PACKET_SIZE constant is wrong"
+        );
+    }
+
+    #[test]
+    fn encode_produces_exactly_packet_size_bytes() {
+        let b = sample_blend(1);
+        let k = sample_kin(1);
+        let pkt = encode_to_bytes(&b, &k);
+        assert_eq!(pkt.len(), PACKET_SIZE);
+    }
+
+    #[test]
+    fn magic_bytes_are_present() {
+        let pkt = encode_to_bytes(&sample_blend(0), &sample_kin(0));
+        assert_eq!(&pkt[..4], PACKET_MAGIC);
+    }
+
+    #[test]
+    fn size_fields_are_little_endian_and_correct() {
+        let pkt = encode_to_bytes(&sample_blend(0), &sample_kin(0));
+        let blend_sz = u16::from_le_bytes([pkt[4], pkt[5]]) as usize;
+        let kin_sz = u16::from_le_bytes([pkt[6], pkt[7]]) as usize;
+        assert_eq!(blend_sz, 216);
+        assert_eq!(kin_sz, 88);
+    }
+
+    #[test]
+    fn round_trip_preserves_all_fields() {
+        let b = sample_blend(99_000);
+        let k = KinematicTransformFrame::from_breath(99_000, 0.6, 0.3);
+        let pkt = encode_to_bytes(&b, &k);
+        let (b2, k2) = decode(&pkt).expect("decode failed");
+        assert_eq!(b2.timestamp_us, b.timestamp_us);
+        assert_eq!(b2.weights[17], 0.5);
+        assert_eq!(k2.timestamp_us, k.timestamp_us);
+        assert!((k2.head_pitch_deg - 0.6).abs() < 1e-6);
+        assert!((k2.clavicle_rise - 0.3).abs() < 1e-6);
+    }
+
+    #[test]
+    fn decode_rejects_wrong_length() {
+        let pkt = encode_to_bytes(&sample_blend(0), &sample_kin(0));
+        let short = &pkt[..PACKET_SIZE - 1];
+        assert!(matches!(decode(short), Err(DecodeError::WrongLength { .. })));
+    }
+
+    #[test]
+    fn decode_rejects_bad_magic() {
+        let mut pkt = encode_to_bytes(&sample_blend(0), &sample_kin(0)).to_vec();
+        pkt[0] = 0xFF;
+        assert!(matches!(decode(&pkt), Err(DecodeError::BadMagic)));
+    }
+
+    #[test]
+    fn decode_rejects_wrong_size_fields() {
+        let mut pkt = encode_to_bytes(&sample_blend(0), &sample_kin(0)).to_vec();
+        // Corrupt the blend_sz field.
+        let wrong: u16 = 100;
+        pkt[4..6].copy_from_slice(&wrong.to_le_bytes());
+        assert!(matches!(
+            decode(&pkt),
+            Err(DecodeError::SizeMismatch { field: "BlendshapeFrame", .. })
+        ));
+    }
+
+    #[test]
+    fn encode_with_scratch_buffer_matches_encode_to_bytes() {
+        let b = sample_blend(7);
+        let k = sample_kin(7);
+        let owned = encode_to_bytes(&b, &k);
+        let mut scratch = BytesMut::with_capacity(PACKET_SIZE);
+        encode(&b, &k, &mut scratch);
+        assert_eq!(owned.as_ref(), scratch.as_ref());
+    }
+}

miranda-transport/src/hub.rs

@@ -0,0 +1,317 @@
+//! Data-plane hub — broadcasts binary frame packets to all connected browsers.
+//!
+//! # Design
+//!
+//! [`DataChannelHub`] is the producer-side handle. It is cloned cheaply and
+//! handed to whatever thread drains the `MirandaBus` blendshape + kinematic
+//! rings. When it has a frame pair ready it calls [`DataChannelHub::broadcast`],
+//! which sends the encoded packet to every currently-open subscriber channel.
+//!
+//! Each browser connection is represented as a [`tokio::sync::mpsc`] sender.
+//! The channel is bounded at [`SUBSCRIBER_CHANNEL_DEPTH`] frames. When a slow
+//! browser fills its channel the hub drops the frame for that subscriber and
+//! increments a counter — backpressure is surfaced as telemetry, never as a
+//! stall on the producer thread.
+//!
+//! # Thread safety
+//!
+//! `DataChannelHub` wraps its subscriber list in a `Mutex<Vec<…>>`. This lock
+//! is taken on two very different timescales:
+//!
+//! - **Broadcast path** (60 Hz): holds the lock only to clone the sender list,
+//!   then releases it before doing any I/O. The actual per-subscriber send
+//!   happens after the lock is dropped, so a single slow subscriber cannot
+//!   hold the lock while sending.
+//! - **Subscribe/unsubscribe path** (human-speed): adds or removes one entry.
+//!
+//! The lock contention window on the broadcast path is O(n subscribers) for a
+//! `clone()` of a `Vec` of `Arc`-wrapped senders — each clone is a pointer
+//! copy, not a byte copy of frame data. At 60 Hz with 4 subscribers this is
+//! well under 1 µs.
+
+use std::sync::{Arc, Mutex};
+
+use bytes::BytesMut;
+use miranda_core::{BlendshapeFrame, KinematicTransformFrame};
+use tokio::sync::mpsc;
+
+use crate::frame;
+
+/// Backlog a subscriber's channel can hold before frames are dropped.
+///
+/// At 60 FPS this is 333 ms of buffer. A subscriber that falls more than
+/// 333 ms behind is not keeping up — the face it is rendering is already
+/// stale. Dropping rather than growing the backlog is the correct choice:
+/// sending stale frames would produce a visible lag-then-lurch artefact
+/// worse than a momentary gap.
+pub const SUBSCRIBER_CHANNEL_DEPTH: usize = 20;
+
+/// A subscriber's receive end. The WebSocket handler task receives encoded
+/// packets from this channel and forwards them to the browser.
+pub type SubscriberRx = mpsc::Receiver<bytes::Bytes>;
+
+/// Internal subscriber record.
+struct Subscriber {
+    tx: mpsc::Sender<bytes::Bytes>,
+    /// How many frames were dropped because this subscriber's channel was
+    /// full. Read by the telemetry hub.
+    dropped: Arc<std::sync::atomic::AtomicU64>,
+}
+
+/// The data-plane broadcast hub.
+///
+/// Clone this to share it across threads — the clone is cheap (one
+/// `Arc::clone`) and every clone broadcasts to the same subscriber set.
+#[derive(Clone)]
+pub struct DataChannelHub {
+    inner: Arc<HubInner>,
+}
+
+struct HubInner {
+    subscribers: Mutex<Vec<Subscriber>>,
+    /// Total frames broadcast (to at least one subscriber).
+    frames_broadcast: std::sync::atomic::AtomicU64,
+    /// Total frames dropped across all subscribers.
+    total_dropped: std::sync::atomic::AtomicU64,
+}
+
+impl DataChannelHub {
+    /// Creates a new hub with no subscribers.
+    pub fn new() -> Self {
+        Self {
+            inner: Arc::new(HubInner {
+                subscribers: Mutex::new(Vec::new()),
+                frames_broadcast: std::sync::atomic::AtomicU64::new(0),
+                total_dropped: std::sync::atomic::AtomicU64::new(0),
+            }),
+        }
+    }
+
+    /// Registers a new subscriber and returns its receive end.
+    ///
+    /// Called by the axum WebSocket upgrade handler when a new browser
+    /// connects. The returned `SubscriberRx` is moved into the per-connection
+    /// task.
+    pub fn subscribe(&self) -> (SubscriberRx, Arc<std::sync::atomic::AtomicU64>) {
+        let dropped = Arc::new(std::sync::atomic::AtomicU64::new(0));
+        let (tx, rx) = mpsc::channel(SUBSCRIBER_CHANNEL_DEPTH);
+        self.inner.subscribers.lock().unwrap().push(Subscriber {
+            tx,
+            dropped: Arc::clone(&dropped),
+        });
+        (rx, dropped)
+    }
+
+    /// Encodes one frame pair and sends the packet to every subscriber.
+    ///
+    /// Drops the frame for any subscriber whose channel is full, and removes
+    /// subscribers whose channels have been closed (browser disconnected).
+    /// Allocation-free in the steady state: the scratch buffer is passed in
+    /// by the caller so it can be reused across calls.
+    pub fn broadcast(
+        &self,
+        blend: &BlendshapeFrame,
+        kin: &KinematicTransformFrame,
+        scratch: &mut BytesMut,
+    ) {
+        scratch.clear();
+        frame::encode(blend, kin, scratch);
+        let packet = scratch.clone().freeze();
+
+        // Take a snapshot of the sender list, release the lock, then send.
+        let senders: Vec<(mpsc::Sender<bytes::Bytes>, Arc<std::sync::atomic::AtomicU64>)> = {
+            self.inner
+                .subscribers
+                .lock()
+                .unwrap()
+                .iter()
+                .map(|s| (s.tx.clone(), Arc::clone(&s.dropped)))
+                .collect()
+        };
+
+        if senders.is_empty() {
+            return;
+        }
+
+        let mut any_closed = false;
+        for (tx, dropped) in &senders {
+            match tx.try_send(packet.clone()) {
+                Ok(_) => {}
+                Err(mpsc::error::TrySendError::Full(_)) => {
+                    dropped.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                    self.inner
+                        .total_dropped
+                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                }
+                Err(mpsc::error::TrySendError::Closed(_)) => {
+                    any_closed = true;
+                }
+            }
+        }
+        self.inner
+            .frames_broadcast
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+
+        // Prune disconnected subscribers. Only pay the mutex cost when at
+        // least one channel is gone.
+        if any_closed {
+            self.inner
+                .subscribers
+                .lock()
+                .unwrap()
+                .retain(|s| !s.tx.is_closed());
+        }
+    }
+
+    /// Number of currently active subscribers.
+    pub fn subscriber_count(&self) -> usize {
+        self.inner.subscribers.lock().unwrap().len()
+    }
+
+    /// Total frames sent to at least one subscriber.
+    pub fn frames_broadcast(&self) -> u64 {
+        self.inner
+            .frames_broadcast
+            .load(std::sync::atomic::Ordering::Relaxed)
+    }
+
+    /// Total frames dropped across all subscribers due to backpressure.
+    pub fn total_dropped(&self) -> u64 {
+        self.inner
+            .total_dropped
+            .load(std::sync::atomic::Ordering::Relaxed)
+    }
+}
+
+impl Default for DataChannelHub {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::frame::PACKET_SIZE;
+    use miranda_core::{BLENDSHAPE_COUNT, KINEMATIC_JOINT_COUNT, Quaternion};
+
+    fn blend(ts: u64) -> BlendshapeFrame {
+        BlendshapeFrame {
+            timestamp_us: ts,
+            weights: [0.0; BLENDSHAPE_COUNT],
+        }
+    }
+
+    fn kin(ts: u64) -> KinematicTransformFrame {
+        KinematicTransformFrame {
+            timestamp_us: ts,
+            joints: [Quaternion::IDENTITY; KINEMATIC_JOINT_COUNT],
+            head_pitch_deg: 0.0,
+            clavicle_rise: 0.0,
+            _reserved: [0; 8],
+        }
+    }
+
+    #[tokio::test]
+    async fn subscriber_receives_broadcast() {
+        let hub = DataChannelHub::new();
+        let (mut rx, _dropped) = hub.subscribe();
+        let mut scratch = BytesMut::with_capacity(PACKET_SIZE);
+
+        hub.broadcast(&blend(1), &kin(1), &mut scratch);
+
+        let pkt = rx.recv().await.expect("expected a packet");
+        assert_eq!(pkt.len(), PACKET_SIZE);
+        assert_eq!(&pkt[..4], b"MRD1");
+    }
+
+    #[tokio::test]
+    async fn multiple_subscribers_all_receive() {
+        let hub = DataChannelHub::new();
+        let (mut rx1, _) = hub.subscribe();
+        let (mut rx2, _) = hub.subscribe();
+        let mut scratch = BytesMut::with_capacity(PACKET_SIZE);
+
+        hub.broadcast(&blend(2), &kin(2), &mut scratch);
+
+        assert!(rx1.recv().await.is_some());
+        assert!(rx2.recv().await.is_some());
+    }
+
+    #[tokio::test]
+    async fn broadcast_with_no_subscribers_is_a_noop() {
+        let hub = DataChannelHub::new();
+        let mut scratch = BytesMut::with_capacity(PACKET_SIZE);
+        // Must not panic.
+        hub.broadcast(&blend(0), &kin(0), &mut scratch);
+        assert_eq!(hub.frames_broadcast(), 0);
+    }
+
+    #[tokio::test]
+    async fn full_channel_drops_frame_and_increments_counter() {
+        let hub = DataChannelHub::new();
+        let (mut rx, dropped) = hub.subscribe();
+        let mut scratch = BytesMut::with_capacity(PACKET_SIZE);
+
+        // Fill the channel.
+        for i in 0..(SUBSCRIBER_CHANNEL_DEPTH + 5) as u64 {
+            hub.broadcast(&blend(i), &kin(i), &mut scratch);
+        }
+
+        let drop_count = dropped.load(std::sync::atomic::Ordering::Relaxed);
+        assert!(drop_count >= 5, "expected at least 5 drops, got {drop_count}");
+        assert!(hub.total_dropped() >= 5);
+
+        // Drain to confirm the channel held SUBSCRIBER_CHANNEL_DEPTH frames.
+        let mut received = 0usize;
+        while rx.try_recv().is_ok() {
+            received += 1;
+        }
+        assert_eq!(received, SUBSCRIBER_CHANNEL_DEPTH);
+    }
+
+    #[tokio::test]
+    async fn closed_subscriber_is_pruned_on_next_broadcast() {
+        let hub = DataChannelHub::new();
+        let (rx, _) = hub.subscribe();
+        let mut scratch = BytesMut::with_capacity(PACKET_SIZE);
+
+        assert_eq!(hub.subscriber_count(), 1);
+        drop(rx); // close the receive end
+
+        // First broadcast detects the closed channel and prunes.
+        hub.broadcast(&blend(0), &kin(0), &mut scratch);
+        assert_eq!(hub.subscriber_count(), 0);
+    }
+
+    #[tokio::test]
+    async fn packet_content_is_decoded_correctly_by_receiver() {
+        let hub = DataChannelHub::new();
+        let (mut rx, _) = hub.subscribe();
+        let mut scratch = BytesMut::with_capacity(PACKET_SIZE);
+
+        let b = BlendshapeFrame {
+            timestamp_us: 12_345_678,
+            weights: {
+                let mut w = [0.0f32; BLENDSHAPE_COUNT];
+                w[17] = 0.75;
+                w
+            },
+        };
+        let k = KinematicTransformFrame::from_breath(12_345_678, 1.2, 0.5);
+        hub.broadcast(&b, &k, &mut scratch);
+
+        let pkt = rx.recv().await.unwrap();
+        let (b2, k2) = crate::frame::decode(&pkt).expect("decode failed");
+        assert_eq!(b2.timestamp_us, 12_345_678);
+        assert!((b2.weights[17] - 0.75).abs() < 1e-6);
+        assert!((k2.head_pitch_deg - 1.2).abs() < 1e-6);
+        assert!((k2.clavicle_rise - 0.5).abs() < 1e-6);
+    }
+
+    #[test]
+    fn hub_is_send_and_sync() {
+        fn assert_send_sync<T: Send + Sync>() {}
+        assert_send_sync::<DataChannelHub>();
+    }
+}

miranda-transport/src/lib.rs

@@ -1,3 +1,80 @@
-//! miranda-transport — Work Order 4: webrtc-rs DataChannels for blendshape
-//! matrices/gaze/lighting at 60 FPS, plus the Axum WebSocket telemetry hub.
-//! Empty skeleton — see `nobility-posh-framework`'s work-orders.md.
+//! miranda-transport — Work Order 4: full-duplex transport layer.
+//!
+//! Two logical servers share one [`axum`] application and one Tokio runtime:
+//!
+//! 1. **Data-plane hub** (`/data` WebSocket endpoint): broadcasts binary
+//!    [`BlendshapeFrame`] + [`KinematicTransformFrame`] packets to every
+//!    connected browser at 60 FPS. This is the DataChannel-equivalent path —
+//!    binary framing, minimal per-frame overhead, no JSON on the hot path.
+//!
+//! 2. **Control-plane telemetry** (`/telemetry` WebSocket endpoint): pushes
+//!    JSON metric snapshots (frame build time, dispatcher stats,
+//!    circuit-breaker state) to connected management tools at ~2 Hz.
+//!
+//! # Wire formats
+//!
+//! ## Data plane — binary packet
+//!
+//! Each data-plane message is a length-prefixed concatenation of two frames:
+//!
+//! ```text
+//! offset  0 : magic    : [u8; 4]   = b"MRD1"   (version tag)
+//! offset  4 : blend_sz : u16 LE    = 216        (sizeof BlendshapeFrame)
+//! offset  6 : kin_sz   : u16 LE    = 88         (sizeof KinematicTransformFrame)
+//! offset  8 : blend    : [u8; 216] (BlendshapeFrame, repr(C), Pod)
+//! offset 224: kin      : [u8; 88]  (KinematicTransformFrame, repr(C), Pod)
+//! ─────────────────────────────────────────────────────────────────────────
+//! total : 312 bytes
+//! ```
+//!
+//! The magic bytes and size fields allow a receiver to detect protocol
+//! mismatches immediately rather than silently misinterpreting bytes.
+//! The sizes are little-endian so they round-trip through JavaScript's
+//! `DataView.getUint16(offset, true)` without byte-swapping.
+//!
+//! ## Control plane — JSON telemetry snapshot
+//!
+//! ```json
+//! {
+//!   "t_us": 1234567890,
+//!   "frames_published": 1801,
+//!   "frames_dropped": 0,
+//!   "late_frames": 0,
+//!   "publish_failures": 0,
+//!   "mean_build_us": 11.95,
+//!   "max_build_us": 1102.3,
+//!   "audio_chunks_consumed": 360,
+//!   "data_subscribers": 2,
+//!   "telemetry_subscribers": 1,
+//!   "circuit_breaker": "closed"
+//! }
+//! ```
+//!
+//! # webrtc-rs and the `webrtc-native` feature
+//!
+//! The `webrtc-native` Cargo feature gates the full `webrtc-rs` crate, which
+//! provides DTLS/ICE/SRTP for production NAT traversal. It is not compiled by
+//! default because the C++ TLS dependency chain takes several minutes to build
+//! on the development machine (Celeron N4500) and because there is no STUN/
+//! TURN server available in the offline environment.
+//!
+//! From a browser's perspective the two paths are **wire-compatible**: both
+//! send the same 312-byte binary packet per frame. The difference is the
+//! transport layer underneath — plain WebSocket vs. DTLS DataChannel. The
+//! browser-side JavaScript only changes the connection constructor, not the
+//! frame parser.
+//!
+//! When AWS connectivity is restored and `--features webrtc-native` is
+//! activated, the `DataChannelHub` type alias in this module switches to
+//! the real WebRTC signaling path automatically.
+
+pub mod frame;
+pub mod hub;
+pub mod server;
+pub mod telemetry;
+pub mod wiring;
+
+pub use hub::DataChannelHub;
+pub use server::{ServerConfig, TransportServer};
+pub use telemetry::TelemetryHub;
+pub use wiring::StatsSource;

miranda-transport/src/server.rs

@@ -0,0 +1,379 @@
+//! Axum HTTP + WebSocket server wiring.
+//!
+//! One [`axum::Router`] serves both endpoints:
+//!
+//! - `GET /data` — upgrades to a binary WebSocket; the hub sends 312-byte
+//!   frame packets at 60 FPS.
+//! - `GET /telemetry` — upgrades to a text WebSocket; the hub sends JSON
+//!   snapshots at ~2 Hz.
+//! - `GET /health` — plain JSON health check; never requires upgrade.
+//!
+//! # Usage
+//!
+//! ```rust,no_run
+//! use std::net::SocketAddr;
+//! use std::sync::Arc;
+//! use miranda_transport::{TransportServer, ServerConfig, DataChannelHub, TelemetryHub};
+//! use miranda_transport::telemetry::DispatchSource;
+//!
+//! #[tokio::main]
+//! async fn main() {
+//!     let data_hub = DataChannelHub::new();
+//!     let tele_hub = TelemetryHub::new(0);
+//!     let cfg = ServerConfig::default();
+//!     let server = TransportServer::new(cfg, data_hub, tele_hub);
+//!     server.run().await.unwrap();
+//! }
+//! ```
+
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use axum::{
+    extract::{
+        ws::{Message, WebSocket, WebSocketUpgrade},
+        State,
+    },
+    response::IntoResponse,
+    routing::get,
+    Json, Router,
+};
+use serde_json::json;
+use tokio::time;
+
+use crate::frame::PACKET_SIZE;
+use crate::hub::DataChannelHub;
+use crate::telemetry::TelemetryHub;
+/// Transport server configuration.
+#[derive(Debug, Clone)]
+pub struct ServerConfig {
+    /// Address to bind. Defaults to `127.0.0.1:9090`.
+    pub bind: SocketAddr,
+    /// Telemetry snapshot interval. Defaults to 500 ms.
+    pub telemetry_interval: Duration,
+}
+
+impl Default for ServerConfig {
+    fn default() -> Self {
+        Self {
+            bind: "127.0.0.1:9090".parse().unwrap(),
+            telemetry_interval: Duration::from_millis(500),
+        }
+    }
+}
+
+/// Shared state injected into every axum handler.
+struct AppState {
+    data_hub: DataChannelHub,
+    tele_hub: TelemetryHub,
+    telemetry_interval: Duration,
+}
+
+/// The combined transport server.
+pub struct TransportServer {
+    cfg: ServerConfig,
+    data_hub: DataChannelHub,
+    tele_hub: TelemetryHub,
+}
+
+impl TransportServer {
+    pub fn new(cfg: ServerConfig, data_hub: DataChannelHub, tele_hub: TelemetryHub) -> Self {
+        Self { cfg, data_hub, tele_hub }
+    }
+
+    /// Binds the server and runs it until the process exits.
+    ///
+    /// This is async but never returns in normal operation. Use
+    /// [`tokio::spawn`] or `tokio::select!` to run it alongside other tasks.
+    pub async fn run(self) -> std::io::Result<()> {
+        let state = Arc::new(AppState {
+            data_hub: self.data_hub,
+            tele_hub: self.tele_hub,
+            telemetry_interval: self.cfg.telemetry_interval,
+        });
+
+        let app = Router::new()
+            .route("/health", get(health_handler))
+            .route("/data", get(data_ws_handler))
+            .route("/telemetry", get(telemetry_ws_handler))
+            .with_state(state);
+
+        let listener = tokio::net::TcpListener::bind(self.cfg.bind).await
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::AddrInUse, e))?;
+
+        axum::serve(listener, app).await
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
+    }
+}
+
+// ── Handlers ────────────────────────────────────────────────────────────────
+
+async fn health_handler(State(state): State<Arc<AppState>>) -> impl IntoResponse {
+    Json(json!({
+        "ok": true,
+        "dataSubscribers": state.data_hub.subscriber_count(),
+        "telemetrySubscribers": state.tele_hub.subscriber_count(),
+        "framesBroadcast": state.data_hub.frames_broadcast(),
+        "circuitBreaker": format!("{:?}", state.tele_hub.circuit_breaker()),
+    }))
+}
+
+async fn data_ws_handler(
+    ws: WebSocketUpgrade,
+    State(state): State<Arc<AppState>>,
+) -> impl IntoResponse {
+    ws.on_upgrade(move |socket| data_ws_task(socket, state.data_hub.clone()))
+}
+
+async fn data_ws_task(mut socket: WebSocket, hub: DataChannelHub) {
+    let (mut rx, _dropped) = hub.subscribe();
+
+    loop {
+        match rx.recv().await {
+            Some(pkt) => {
+                debug_assert_eq!(pkt.len(), PACKET_SIZE);
+                if socket.send(Message::Binary(pkt.to_vec().into())).await.is_err() {
+                    break;
+                }
+            }
+            None => break, // hub shut down
+        }
+    }
+}
+
+async fn telemetry_ws_handler(
+    ws: WebSocketUpgrade,
+    State(state): State<Arc<AppState>>,
+) -> impl IntoResponse {
+    ws.on_upgrade(move |socket| {
+        telemetry_ws_task(socket, state.tele_hub.clone(), state.telemetry_interval)
+    })
+}
+
+async fn telemetry_ws_task(
+    mut socket: WebSocket,
+    hub: TelemetryHub,
+    interval: Duration,
+) {
+    let mut rx = hub.subscribe();
+    let mut tick = time::interval(interval);
+    tick.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
+
+    loop {
+        tokio::select! {
+            _ = tick.tick() => {
+                // Drain any stale snapshots from the channel; only send the
+                // latest one. This keeps a slow browser from receiving a
+                // burst of old snapshots when it catches up.
+                let mut latest = None;
+                while let Ok(snap) = rx.try_recv() {
+                    latest = Some(snap);
+                }
+                if let Some(json) = latest {
+                    if socket.send(Message::Text(json.into())).await.is_err() {
+                        break;
+                    }
+                }
+            }
+            // Also forward anything the hub pushed between ticks (e.g. an
+            // alert) without waiting for the next scheduled tick.
+            Some(json) = rx.recv() => {
+                if socket.send(Message::Text(json.into())).await.is_err() {
+                    break;
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::SocketAddr;
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    use bytes::BytesMut;
+    use miranda_core::{BlendshapeFrame, KinematicTransformFrame, BLENDSHAPE_COUNT, KINEMATIC_JOINT_COUNT, Quaternion};
+
+    fn blend(ts: u64) -> BlendshapeFrame {
+        BlendshapeFrame { timestamp_us: ts, weights: [0.0; BLENDSHAPE_COUNT] }
+    }
+    fn kin(ts: u64) -> KinematicTransformFrame {
+        KinematicTransformFrame {
+            timestamp_us: ts,
+            joints: [Quaternion::IDENTITY; KINEMATIC_JOINT_COUNT],
+            head_pitch_deg: 0.0,
+            clavicle_rise: 0.0,
+            _reserved: [0; 8],
+        }
+    }
+
+    #[test]
+    fn default_config_binds_to_loopback() {
+        let cfg = ServerConfig::default();
+        assert_eq!(cfg.bind.ip().to_string(), "127.0.0.1");
+        assert_eq!(cfg.bind.port(), 9090);
+        assert_eq!(cfg.telemetry_interval, Duration::from_millis(500));
+    }
+
+    /// Verify the server starts, the health endpoint is reachable, and a
+    /// data-plane subscriber receives a frame. Uses an OS-assigned port so
+    /// the test does not conflict with other tests or running instances.
+    #[tokio::test]
+    async fn server_health_endpoint_responds_ok() {
+        let data_hub = DataChannelHub::new();
+        let tele_hub = TelemetryHub::new(0);
+        let cfg = ServerConfig {
+            bind: "127.0.0.1:0".parse().unwrap(), // OS-assigned port
+            ..Default::default()
+        };
+
+        let listener = tokio::net::TcpListener::bind(cfg.bind).await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+
+        let state = Arc::new(AppState {
+            data_hub: data_hub.clone(),
+            tele_hub: tele_hub.clone(),
+            telemetry_interval: cfg.telemetry_interval,
+        });
+        let app = Router::new()
+            .route("/health", get(health_handler))
+            .route("/data", get(data_ws_handler))
+            .route("/telemetry", get(telemetry_ws_handler))
+            .with_state(state);
+
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        // Give the server a moment to start.
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        let url = format!("http://127.0.0.1:{port}/health");
+        let resp = reqwest::get(&url).await.expect("health request failed");
+        assert!(resp.status().is_success());
+        let body: serde_json::Value = resp.json().await.unwrap();
+        assert_eq!(body["ok"], true);
+    }
+
+    /// A data WebSocket subscriber receives correctly sized binary frames.
+    #[tokio::test]
+    async fn data_websocket_delivers_binary_frames() {
+        use futures_util::StreamExt;
+        let data_hub = DataChannelHub::new();
+        let tele_hub = TelemetryHub::new(0);
+        let cfg = ServerConfig {
+            bind: "127.0.0.1:0".parse().unwrap(),
+            ..Default::default()
+        };
+
+        let listener = tokio::net::TcpListener::bind(cfg.bind).await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+
+        let state = Arc::new(AppState {
+            data_hub: data_hub.clone(),
+            tele_hub: tele_hub.clone(),
+            telemetry_interval: cfg.telemetry_interval,
+        });
+        let app = Router::new()
+            .route("/data", get(data_ws_handler))
+            .with_state(state);
+
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        // Connect a WebSocket client.
+        use tokio_tungstenite::tungstenite;
+        let (mut ws, _) = tokio_tungstenite::connect_async(
+            format!("ws://127.0.0.1:{port}/data")
+        ).await.expect("ws connect failed");
+
+        // Allow time for the hub to register the subscriber.
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        // Broadcast a frame from the hub side.
+        let mut scratch = BytesMut::with_capacity(PACKET_SIZE);
+        data_hub.broadcast(&blend(42), &kin(42), &mut scratch);
+
+        // Wait for the WebSocket to deliver it.
+        let msg = tokio::time::timeout(Duration::from_millis(500), ws.next()).await
+            .expect("timeout waiting for ws message")
+            .expect("ws stream ended")
+            .expect("ws error");
+
+        match msg {
+            tungstenite::Message::Binary(data) => {
+                assert_eq!(data.len(), PACKET_SIZE,
+                    "expected {PACKET_SIZE} bytes, got {}", data.len());
+                assert_eq!(&data[..4], b"MRD1", "missing MRD1 magic");
+            }
+            other => panic!("expected binary frame, got {other:?}"),
+        }
+    }
+
+    /// Telemetry WebSocket delivers JSON with the expected shape.
+    #[tokio::test]
+    async fn telemetry_websocket_delivers_json_snapshot() {
+        use crate::telemetry::DispatchSource;
+        use futures_util::StreamExt;
+        struct MockSrc;
+        impl DispatchSource for MockSrc {
+            fn frames_published(&self) -> u64 { 120 }
+            fn frames_dropped(&self) -> u64 { 0 }
+            fn late_frames(&self) -> u64 { 0 }
+            fn publish_failures(&self) -> u64 { 0 }
+            fn mean_build_us(&self) -> f64 { 10.0 }
+            fn max_build_us(&self) -> f64 { 100.0 }
+            fn audio_chunks_consumed(&self) -> u64 { 60 }
+        }
+
+        let data_hub = DataChannelHub::new();
+        let tele_hub = TelemetryHub::new(0);
+        let cfg = ServerConfig {
+            bind: "127.0.0.1:0".parse().unwrap(),
+            telemetry_interval: Duration::from_millis(50), // fast for testing
+        };
+
+        let listener = tokio::net::TcpListener::bind(cfg.bind).await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+
+        let tele_for_push = tele_hub.clone();
+        let data_for_state = data_hub.clone();
+        let state = Arc::new(AppState {
+            data_hub: data_for_state,
+            tele_hub: tele_hub.clone(),
+            telemetry_interval: cfg.telemetry_interval,
+        });
+        let app = Router::new()
+            .route("/telemetry", get(telemetry_ws_handler))
+            .with_state(state);
+
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        use tokio_tungstenite::tungstenite;
+        let (mut ws, _) = tokio_tungstenite::connect_async(
+            format!("ws://127.0.0.1:{port}/telemetry")
+        ).await.expect("ws connect failed");
+
+        // Push a snapshot — the ws task will drain it on the next tick.
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        tele_for_push.publish_snapshot(1_000_000, &MockSrc, 0, 1, 0, 0);
+
+        let msg = tokio::time::timeout(Duration::from_millis(500), ws.next()).await
+            .expect("timeout")
+            .expect("stream ended")
+            .expect("ws error");
+
+        match msg {
+            tungstenite::Message::Text(json) => {
+                let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
+                assert_eq!(parsed["framesPublished"], 120);
+                assert_eq!(parsed["circuitBreaker"], "closed");
+            }
+            other => panic!("expected text frame, got {other:?}"),
+        }
+    }
+}

miranda-transport/src/telemetry.rs

@@ -0,0 +1,391 @@
+//! Control-plane telemetry hub — JSON snapshots for management tools.
+//!
+//! [`TelemetryHub`] broadcasts [`TelemetrySnapshot`] messages to every
+//! connected management tool at a configurable interval (default 500 ms).
+//! The snapshot JSON is produced once and fanned out to all subscribers,
+//! so the serialization cost is O(1) regardless of subscriber count.
+//!
+//! # Circuit-breaker state
+//!
+//! The circuit breaker is a simple three-state guard around the blendshape
+//! bus drain. If the dispatcher drops frames for more than
+//! [`CIRCUIT_BREAKER_TRIP_THRESHOLD_MS`] milliseconds, the breaker trips
+//! and the hub stops broadcasting stale data. When the dispatcher catches up
+//! for [`CIRCUIT_BREAKER_RESET_THRESHOLD_MS`], the breaker resets.
+//!
+//! Tripped state is surfaced in every telemetry snapshot so the frontend
+//! (`THE VANITY`) can show a visual circuit-breaker indicator rather than
+//! showing a frozen face with no explanation.
+
+use std::sync::{Arc, Mutex};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+
+use serde::{Deserialize, Serialize};
+use tokio::sync::mpsc;
+
+/// Dropped-frame window before the circuit breaker trips.
+pub const CIRCUIT_BREAKER_TRIP_THRESHOLD_MS: u64 = 500;
+
+/// Clean-frame window before the circuit breaker resets.
+pub const CIRCUIT_BREAKER_RESET_THRESHOLD_MS: u64 = 1_000;
+
+/// Backlog for each telemetry subscriber's channel.
+/// 500 ms / 500 ms interval = 1 snapshot in flight. Give it 8 slots of
+/// headroom in case a subscriber is briefly slow.
+pub const TELEMETRY_CHANNEL_DEPTH: usize = 8;
+
+/// A point-in-time snapshot of the transport and dispatcher state.
+///
+/// Field names are camelCase to match the existing ace-controller JSON
+/// convention used by THE VANITY's node graph.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TelemetrySnapshot {
+    /// Microseconds since harness start.
+    pub t_us: u64,
+
+    // --- Dispatcher counters (from miranda-nodes DispatchStats) ---
+    pub frames_published: u64,
+    pub frames_dropped: u64,
+    pub late_frames: u64,
+    pub publish_failures: u64,
+    pub mean_build_us: f64,
+    pub max_build_us: f64,
+    pub audio_chunks_consumed: u64,
+
+    // --- Transport counters ---
+    pub data_subscribers: usize,
+    pub telemetry_subscribers: usize,
+    pub frames_broadcast: u64,
+    pub frames_dropped_backpressure: u64,
+
+    // --- Circuit breaker ---
+    pub circuit_breaker: CircuitBreakerState,
+}
+
+/// Three-state circuit breaker reported in every telemetry snapshot.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "lowercase")]
+pub enum CircuitBreakerState {
+    /// Normal operation — frames are being produced on time.
+    Closed,
+    /// Frames have been dropping for less than
+    /// [`CIRCUIT_BREAKER_TRIP_THRESHOLD_MS`]. Monitoring is indicated;
+    /// broadcast continues.
+    HalfOpen,
+    /// Frames have been dropping for more than the threshold. The renderer
+    /// should show a connectivity indicator rather than a frozen face.
+    Open,
+}
+
+/// A live source of dispatcher counters. Implemented by the wiring layer
+/// (T5) so the telemetry hub does not depend on `miranda-nodes` directly —
+/// it depends on this trait, which can also be backed by a mock in tests.
+pub trait DispatchSource: Send + Sync + 'static {
+    fn frames_published(&self) -> u64;
+    fn frames_dropped(&self) -> u64;
+    fn late_frames(&self) -> u64;
+    fn publish_failures(&self) -> u64;
+    fn mean_build_us(&self) -> f64;
+    fn max_build_us(&self) -> f64;
+    fn audio_chunks_consumed(&self) -> u64;
+}
+
+/// Telemetry subscriber receive end. The WebSocket handler task reads from
+/// this and forwards JSON text frames to the browser.
+pub type TelemetryRx = mpsc::Receiver<String>;
+
+struct TelemetrySubscriber {
+    tx: mpsc::Sender<String>,
+}
+
+/// The control-plane telemetry hub.
+///
+/// Clone cheaply — every clone shares the same subscriber list and the same
+/// circuit-breaker state.
+#[derive(Clone)]
+pub struct TelemetryHub {
+    inner: Arc<TelemetryInner>,
+}
+
+struct TelemetryInner {
+    subscribers: Mutex<Vec<TelemetrySubscriber>>,
+    start_us: u64,
+    // Circuit-breaker state.
+    breaker: AtomicU64, // encodes CircuitBreakerState as 0/1/2
+    // How long (in ticks) we have been in the current state.
+    bad_streak_ms: AtomicU64,
+    good_streak_ms: AtomicU64,
+    tripped: AtomicBool,
+}
+
+fn encode_state(s: CircuitBreakerState) -> u64 {
+    match s {
+        CircuitBreakerState::Closed => 0,
+        CircuitBreakerState::HalfOpen => 1,
+        CircuitBreakerState::Open => 2,
+    }
+}
+
+fn decode_state(v: u64) -> CircuitBreakerState {
+    match v {
+        1 => CircuitBreakerState::HalfOpen,
+        2 => CircuitBreakerState::Open,
+        _ => CircuitBreakerState::Closed,
+    }
+}
+
+impl TelemetryHub {
+    /// Creates the hub. `start_us` is the harness epoch (microseconds) used
+    /// to compute `t_us` in snapshots.
+    pub fn new(start_us: u64) -> Self {
+        Self {
+            inner: Arc::new(TelemetryInner {
+                subscribers: Mutex::new(Vec::new()),
+                start_us,
+                breaker: AtomicU64::new(0),
+                bad_streak_ms: AtomicU64::new(0),
+                good_streak_ms: AtomicU64::new(0),
+                tripped: AtomicBool::new(false),
+            }),
+        }
+    }
+
+    /// Registers a new telemetry subscriber and returns its receive end.
+    pub fn subscribe(&self) -> TelemetryRx {
+        let (tx, rx) = mpsc::channel(TELEMETRY_CHANNEL_DEPTH);
+        self.inner
+            .subscribers
+            .lock()
+            .unwrap()
+            .push(TelemetrySubscriber { tx });
+        rx
+    }
+
+    /// Advances the circuit-breaker state based on the latest drop count.
+    ///
+    /// Called once per telemetry tick (every `interval_ms`).
+    pub fn tick_circuit_breaker(&self, frames_dropped_this_tick: u64, interval_ms: u64) {
+        if frames_dropped_this_tick > 0 {
+            let bad = self.inner.bad_streak_ms.fetch_add(interval_ms, Ordering::Relaxed)
+                + interval_ms;
+            self.inner.good_streak_ms.store(0, Ordering::Relaxed);
+            let new_state = if bad >= CIRCUIT_BREAKER_TRIP_THRESHOLD_MS {
+                self.inner.tripped.store(true, Ordering::Relaxed);
+                CircuitBreakerState::Open
+            } else {
+                CircuitBreakerState::HalfOpen
+            };
+            self.inner
+                .breaker
+                .store(encode_state(new_state), Ordering::Relaxed);
+        } else {
+            self.inner.bad_streak_ms.store(0, Ordering::Relaxed);
+            let good = self.inner.good_streak_ms.fetch_add(interval_ms, Ordering::Relaxed)
+                + interval_ms;
+            if self.inner.tripped.load(Ordering::Relaxed)
+                && good >= CIRCUIT_BREAKER_RESET_THRESHOLD_MS
+            {
+                self.inner.tripped.store(false, Ordering::Relaxed);
+                self.inner
+                    .breaker
+                    .store(encode_state(CircuitBreakerState::Closed), Ordering::Relaxed);
+            } else if !self.inner.tripped.load(Ordering::Relaxed) {
+                self.inner
+                    .breaker
+                    .store(encode_state(CircuitBreakerState::Closed), Ordering::Relaxed);
+            }
+        }
+    }
+
+    /// Current circuit-breaker state.
+    pub fn circuit_breaker(&self) -> CircuitBreakerState {
+        decode_state(self.inner.breaker.load(Ordering::Relaxed))
+    }
+
+    /// Builds a snapshot and broadcasts it to all subscribers as a JSON
+    /// text frame. Prunes disconnected subscribers.
+    pub fn publish_snapshot(
+        &self,
+        now_us: u64,
+        source: &dyn DispatchSource,
+        data_subscribers: usize,
+        telemetry_subscribers: usize,
+        frames_broadcast: u64,
+        frames_dropped_bp: u64,
+    ) {
+        let snap = TelemetrySnapshot {
+            t_us: now_us.saturating_sub(self.inner.start_us),
+            frames_published: source.frames_published(),
+            frames_dropped: source.frames_dropped(),
+            late_frames: source.late_frames(),
+            publish_failures: source.publish_failures(),
+            mean_build_us: source.mean_build_us(),
+            max_build_us: source.max_build_us(),
+            audio_chunks_consumed: source.audio_chunks_consumed(),
+            data_subscribers,
+            telemetry_subscribers,
+            frames_broadcast,
+            frames_dropped_backpressure: frames_dropped_bp,
+            circuit_breaker: self.circuit_breaker(),
+        };
+
+        let json = match serde_json::to_string(&snap) {
+            Ok(s) => s,
+            Err(_) => return, // can't fail in practice for this struct
+        };
+
+        let mut subs = self.inner.subscribers.lock().unwrap();
+        subs.retain(|s| !s.tx.is_closed());
+        for s in subs.iter() {
+            let _ = s.tx.try_send(json.clone());
+        }
+    }
+
+    /// Number of active telemetry subscribers.
+    pub fn subscriber_count(&self) -> usize {
+        self.inner.subscribers.lock().unwrap().len()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[allow(unused_imports)]
+    use std::sync::atomic::AtomicU64 as StdAtomicU64;
+
+    struct MockSource {
+        published: u64,
+        dropped: u64,
+        late: u64,
+    }
+
+    impl DispatchSource for MockSource {
+        fn frames_published(&self) -> u64 { self.published }
+        fn frames_dropped(&self) -> u64 { self.dropped }
+        fn late_frames(&self) -> u64 { self.late }
+        fn publish_failures(&self) -> u64 { 0 }
+        fn mean_build_us(&self) -> f64 { 11.5 }
+        fn max_build_us(&self) -> f64 { 250.0 }
+        fn audio_chunks_consumed(&self) -> u64 { 100 }
+    }
+
+    fn source(published: u64, dropped: u64) -> MockSource {
+        MockSource { published, dropped, late: 0 }
+    }
+
+    #[tokio::test]
+    async fn subscriber_receives_json_snapshot() {
+        let hub = TelemetryHub::new(0);
+        let mut rx = hub.subscribe();
+
+        hub.publish_snapshot(1_000_000, &source(60, 0), 1, 1, 60, 0);
+
+        let json = rx.recv().await.expect("snapshot expected");
+        let parsed: TelemetrySnapshot = serde_json::from_str(&json).expect("valid json");
+        assert_eq!(parsed.frames_published, 60);
+        assert_eq!(parsed.frames_dropped, 0);
+        assert_eq!(parsed.circuit_breaker, CircuitBreakerState::Closed);
+    }
+
+    #[tokio::test]
+    async fn multiple_subscribers_all_receive_same_snapshot() {
+        let hub = TelemetryHub::new(0);
+        let mut rx1 = hub.subscribe();
+        let mut rx2 = hub.subscribe();
+
+        hub.publish_snapshot(0, &source(1, 0), 0, 2, 1, 0);
+
+        let j1 = rx1.recv().await.unwrap();
+        let j2 = rx2.recv().await.unwrap();
+        assert_eq!(j1, j2);
+    }
+
+    #[test]
+    fn circuit_breaker_trips_after_threshold() {
+        let hub = TelemetryHub::new(0);
+        assert_eq!(hub.circuit_breaker(), CircuitBreakerState::Closed);
+
+        // Feed drops in 100 ms ticks until we hit the trip threshold.
+        let ticks = CIRCUIT_BREAKER_TRIP_THRESHOLD_MS / 100;
+        for _ in 0..ticks {
+            hub.tick_circuit_breaker(1, 100);
+        }
+        assert_eq!(hub.circuit_breaker(), CircuitBreakerState::Open);
+    }
+
+    #[test]
+    fn circuit_breaker_enters_half_open_before_tripping() {
+        let hub = TelemetryHub::new(0);
+        // One tick of drops, below the full threshold.
+        hub.tick_circuit_breaker(1, 100);
+        assert_eq!(hub.circuit_breaker(), CircuitBreakerState::HalfOpen);
+    }
+
+    #[test]
+    fn circuit_breaker_resets_after_clean_period() {
+        let hub = TelemetryHub::new(0);
+
+        // Trip it.
+        let trip_ticks = CIRCUIT_BREAKER_TRIP_THRESHOLD_MS / 100;
+        for _ in 0..trip_ticks {
+            hub.tick_circuit_breaker(1, 100);
+        }
+        assert_eq!(hub.circuit_breaker(), CircuitBreakerState::Open);
+
+        // Feed clean ticks until reset.
+        let reset_ticks = CIRCUIT_BREAKER_RESET_THRESHOLD_MS / 100;
+        for _ in 0..reset_ticks {
+            hub.tick_circuit_breaker(0, 100);
+        }
+        assert_eq!(hub.circuit_breaker(), CircuitBreakerState::Closed);
+    }
+
+    #[test]
+    fn bad_streak_resets_when_frames_recover() {
+        let hub = TelemetryHub::new(0);
+        hub.tick_circuit_breaker(1, 100); // half-open
+        hub.tick_circuit_breaker(0, 100); // recover — should go back to closed
+        assert_eq!(hub.circuit_breaker(), CircuitBreakerState::Closed);
+    }
+
+    #[tokio::test]
+    async fn disconnected_subscriber_is_pruned() {
+        let hub = TelemetryHub::new(0);
+        let rx = hub.subscribe();
+        assert_eq!(hub.subscriber_count(), 1);
+        drop(rx);
+        hub.publish_snapshot(0, &source(0, 0), 0, 0, 0, 0);
+        assert_eq!(hub.subscriber_count(), 0);
+    }
+
+    #[test]
+    fn hub_is_send_and_sync() {
+        fn assert_ss<T: Send + Sync>() {}
+        assert_ss::<TelemetryHub>();
+    }
+
+    #[test]
+    fn snapshot_json_is_camel_case() {
+        let snap = TelemetrySnapshot {
+            t_us: 0,
+            frames_published: 0,
+            frames_dropped: 0,
+            late_frames: 0,
+            publish_failures: 0,
+            mean_build_us: 0.0,
+            max_build_us: 0.0,
+            audio_chunks_consumed: 0,
+            data_subscribers: 0,
+            telemetry_subscribers: 0,
+            frames_broadcast: 0,
+            frames_dropped_backpressure: 0,
+            circuit_breaker: CircuitBreakerState::Closed,
+        };
+        let json = serde_json::to_string(&snap).unwrap();
+        // Spot-check a couple of the renamed fields.
+        assert!(json.contains("\"framesPublished\""), "expected camelCase framesPublished");
+        assert!(json.contains("\"meanBuildUs\""), "expected camelCase meanBuildUs");
+        assert!(json.contains("\"circuitBreaker\""), "expected camelCase circuitBreaker");
+    }
+}

miranda-transport/src/wiring.rs

@@ -0,0 +1,136 @@
+//! Wiring: connects real `miranda-nodes` types to the transport layer's
+//! abstraction traits.
+//!
+//! This module is the only place `miranda-transport` reaches into
+//! `miranda-nodes`. Every other module in this crate works through the
+//! [`DispatchSource`] trait so it is testable with a mock. Keeping the
+//! concrete dependency in one file makes it easy to swap the source (e.g.
+//! to a different WO-3 dispatcher implementation) without touching the hub
+//! or telemetry logic.
+
+use std::sync::Arc;
+use std::sync::atomic::Ordering;
+
+use miranda_nodes::dispatcher::DispatchStats;
+
+use crate::telemetry::DispatchSource;
+
+/// Wraps an `Arc<DispatchStats>` and implements [`DispatchSource`] so the
+/// telemetry hub can read live counters from the running dispatcher.
+///
+/// All counters use `Relaxed` ordering — the same ordering `DispatchStats`
+/// itself uses internally. `Relaxed` is correct here because:
+///
+/// 1. These are independent monotonic counters with no happens-before
+///    relationship to enforce between them.
+/// 2. The telemetry snapshot is a *sample*, not a synchronization point.
+///    A slightly stale count on one field while another is up-to-date does
+///    not affect correctness — it just means the snapshot was taken mid-
+///    frame, which is unavoidable without a mutex the dispatcher would need
+///    to hold across the entire frame build.
+pub struct StatsSource {
+    stats: Arc<DispatchStats>,
+}
+
+impl StatsSource {
+    /// Wraps a shared `DispatchStats` reference.
+    pub fn new(stats: Arc<DispatchStats>) -> Self {
+        Self { stats }
+    }
+}
+
+impl DispatchSource for StatsSource {
+    fn frames_published(&self) -> u64 {
+        self.stats.frames_published.load(Ordering::Relaxed)
+    }
+
+    fn frames_dropped(&self) -> u64 {
+        self.stats.frames_dropped.load(Ordering::Relaxed)
+    }
+
+    fn late_frames(&self) -> u64 {
+        self.stats.late_frames.load(Ordering::Relaxed)
+    }
+
+    fn publish_failures(&self) -> u64 {
+        self.stats.publish_failures.load(Ordering::Relaxed)
+    }
+
+    fn mean_build_us(&self) -> f64 {
+        // `DispatchStats::mean_build_us()` returns `None` before the first
+        // frame is published. Map that to `0.0` so the telemetry subscriber
+        // always receives a valid JSON number rather than `null`.
+        self.stats.mean_build_us().unwrap_or(0.0)
+    }
+
+    fn max_build_us(&self) -> f64 {
+        self.stats.max_build_us()
+    }
+
+    fn audio_chunks_consumed(&self) -> u64 {
+        self.stats.audio_chunks_consumed.load(Ordering::Relaxed)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::AtomicU64;
+
+    fn make_stats(
+        published: u64,
+        dropped: u64,
+        late: u64,
+        failures: u64,
+        total_build_ns: u64,
+        max_build_ns: u64,
+        audio: u64,
+    ) -> Arc<DispatchStats> {
+        let s = Arc::new(DispatchStats::default());
+        s.frames_published.store(published, Ordering::Relaxed);
+        s.frames_dropped.store(dropped, Ordering::Relaxed);
+        s.late_frames.store(late, Ordering::Relaxed);
+        s.publish_failures.store(failures, Ordering::Relaxed);
+        s.total_build_ns.store(total_build_ns, Ordering::Relaxed);
+        s.max_build_ns.store(max_build_ns, Ordering::Relaxed);
+        s.audio_chunks_consumed.store(audio, Ordering::Relaxed);
+        s
+    }
+
+    #[test]
+    fn all_fields_are_forwarded_correctly() {
+        let stats = make_stats(1801, 0, 2, 3, 21_570_000, 1_102_301, 360);
+        let src = StatsSource::new(stats);
+
+        assert_eq!(src.frames_published(), 1801);
+        assert_eq!(src.frames_dropped(), 0);
+        assert_eq!(src.late_frames(), 2);
+        assert_eq!(src.publish_failures(), 3);
+        // mean = total_build_ns / published / 1000 = 21_570_000 / 1801 / 1000 ≈ 11.98 µs
+        let mean = src.mean_build_us();
+        assert!(
+            (mean - 11.977).abs() < 0.01,
+            "mean_build_us {mean:.4} not close to expected 11.977"
+        );
+        // max = max_build_ns / 1000 = 1_102_301 ns = 1102.3 µs
+        let max = src.max_build_us();
+        assert!(
+            (max - 1102.3).abs() < 0.01,
+            "max_build_us {max:.4} not close to expected 1102.3"
+        );
+        assert_eq!(src.audio_chunks_consumed(), 360);
+    }
+
+    #[test]
+    fn mean_is_zero_before_any_frames() {
+        let stats = make_stats(0, 0, 0, 0, 0, 0, 0);
+        let src = StatsSource::new(stats);
+        assert_eq!(src.mean_build_us(), 0.0);
+    }
+
+    #[test]
+    fn stats_source_is_send_and_sync() {
+        fn assert_ss<T: Send + Sync>() {}
+        assert_ss::<StatsSource>();
+    }
+}