feat(webrtc): real-SCTP loopback provider + typed msg transforms

[spomichter]

first chunk of the webrtc-as-full-transport-backend work. two peices, can split the transforms commit out if prefered.

loopback provider — keyless real-SCTP link for tests + benchmarks. full aiortc stack (dtls, sctp, localhost udp), no signaling server, no CF creds. unlike the in-memory mock this measures what a real datachannel actually does:

LoopbackConfig(ProviderConfig)              # frozen, picklable, singelton per process
  -> LoopbackProvider(AsyncProviderBase)
       _connect():
         pc_a, pc_b = RTCPeerConnection(iceServers=[])   # empty! aiortc defualt queries google stun -> exactly 5s stall per peer when blocked
         both.createDataChannel("_init", negotiated, id=0)   # forces the sctp m-line pre-offer
         offer/answer exchanged directly in process, host candidates only
       publish(topic, data):
         ch = channels[topic] or ensure_channel(topic)   # lazy negotiated pair, same id on both pcs
         loop.call_soon_threadsafe(ch.send, data)
       subscribe(topic, cb):
         callbacks[topic] += cb                          # recv-side pc dispatches on("message")

benchmarked (new WebrtcSctp row, keyless, runs anywhere): ~3.7k msg/s small msgs, ~20 MiB/s at 16-64KiB, 0% loss once the send queue drains. so raw Image/PointCloud2 over datachannels is dead — decimation / video track routing it is. thats the decision gate this was built to answer.

transforms — decimation/throttling as typed picklable callables that compose around any transport, not webrtc specific:

MsgTransform = (A) -> B | None          # None = drop the message
TransformTransport(inner, *fns)         # broadcast() applies fns in order, then delegates

("lidar", PointCloud2): TransformTransport(ZenohTransport(...), VoxelDownsample(0.05))
("color_image", Image): TransformTransport(WebRTCTransport(...), Throttle(5.0), ResizeImage(640, 480))

tests: 6 loopback units (event driven, no sleeps), real-sctp row in the shared pubsub contract grid, 6 transform units. full suite green locally (2510 passed). loopback tests add ~0.5s to CI.

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
2737	1	2736	73

View the top 1 failed test(s) by shortest run time

dimos.protocol.pubsub.test_spec::test_high_volume_messages[webrtc_sctp_context-test/sctp_topic-values5]

Stack Traces | 0.935s run time

pubsub_context = <function webrtc_sctp_context at 0x702834c1e340>
topic = 'test/sctp_topic'
values = [b'webrtc_sctp_value1', b'webrtc_sctp_value2', b'webrtc_sctp_value3']

    @pytest.mark.self_hosted
    @pytest.mark.skipif_macos_bug
    @pytest.mark.parametrize("pubsub_context, topic, values", testdata)
    def test_high_volume_messages(
        pubsub_context: Callable[[], Any], topic: Any, values: list[Any]
    ) -> None:
        """Test that all 5k messages are received correctly.
        Limited to 5k because ros transport cannot handle more.
        Might want to have separate expectations per transport later
        """
        with pubsub_context() as x:
            # Create a list to capture received messages
            received_messages: list[Any] = []
            last_message_time = [time.time()]  # Use list to allow modification in callback
    
            # Define callback function
            def callback(message: Any, topic: Any) -> None:
                received_messages.append(message)
                last_message_time[0] = time.time()
    
            # Subscribe to the topic
            x.subscribe(topic, callback)
    
            # Publish 5000 messages
            num_messages = 5000
            for _ in range(num_messages):
                x.publish(topic, values[0])
    
            # Wait until no messages received for 0.5 seconds
            timeout = 2.0  # Maximum time to wait
            stable_duration = 0.1  # Time without new messages to consider done
            start_time = time.time()
    
            while time.time() - start_time < timeout:
                if time.time() - last_message_time[0] >= stable_duration:
                    break
                time.sleep(0.1)
    
            # Capture count and clear list to avoid printing huge list on failure
            received_len = len(received_messages)
            received_messages.clear()
>           assert received_len == num_messages, f"Expected {num_messages} messages, got {received_len}"
E           AssertionError: Expected 5000 messages, got 0
E           assert 0 == 5000

_          = 4999
callback   = <function test_high_volume_messages.<locals>.callback at 0x70279d0ad300>
last_message_time = [1783538226.45962]
num_messages = 5000
pubsub_context = <function webrtc_sctp_context at 0x702834c1e340>
received_len = 0
received_messages = [b'webrtc_sctp_value1', b'webrtc_sctp_value1', b'webrtc_sctp_value1', b'webrtc_sctp_value1', b'webrtc_sctp_value1', b'webrtc_sctp_value1', ...]
stable_duration = 0.1
start_time = 1783538224.7404108
timeout    = 2.0
topic      = 'test/sctp_topic'
values     = [b'webrtc_sctp_value1', b'webrtc_sctp_value2', b'webrtc_sctp_value3']
x          = <dimos.protocol.pubsub.impl.webrtc.webrtcpubsub.WebRTCPubSub object at 0x70280618dd60>

.../protocol/pubsub/test_spec.py:408: AssertionError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}