WIP: Paul/feat/t2 relay

[paul-nechifor]

Contribution path

• Small, safe change that does not need a tracking issue
• Linked issue or discussion: DIM-XXX / #XXX / URL

Problem

Solution

How to Test

AI assistance

Checklist

• This PR is scoped to one issue or clearly stated problem.
• I ran the relevant checks (uv run pytest, pre-commit) for the files I changed.
• I have reviewed and understood every line in this PR.
• I disclosed AI assistance above.
• I have read and approved the CLA.

[greptile-apps]

Greptile Summary

This PR adds a WebTransport relay between robot streams and cockpit viewers. The main changes are:

• Robot identity and channel-manifest registration.
• Viewer watch and per-channel subscription routing.
• Lazy relay-bridge encoding and reconnect handling.
• Shared Python and TypeScript protocol extensions.

Confidence Score: 4/5

The viewer subscription path must enforce handshake completion before merging.

• Pre-hello viewer control messages can enable robot streams and receive forwarded frames.
• Relay lifecycle, reconnect, and protocol validation paths otherwise include targeted guards.

web/relay/registry.ts, web/relay/session.ts

Security Review

The viewer handshake state is not enforced before watch, subscription, and frame-routing operations. A connected client can subscribe and receive robot frames without first completing the expected viewer hello.

Important Files Changed

Filename	Overview
web/relay/registry.ts	Adds robot registration, subscription snapshots, and selective frame routing; viewer handshake state is not enforced on routing operations.
web/relay/session.ts	Adds robot and viewer WebTransport session orchestration, including handshake and transport-specific control loops.
dimos/web/relay_bridge/relay_bridge_module.py	Adds lazy stream encoding, local relay management, reconnect supervision, and child-process recovery.
dimos/web/relay_bridge/wt_client.py	Extends the WebTransport client with manifest hello messages, relay controls, bounded datagrams, and idempotent close handling.
web/shared/protocol.ts	Extends the shared relay protocol with robot, manifest, watch, subscription, and snapshot messages.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  Robot[Robot bridge] -->|hello + manifest| RobotSession
  RobotSession --> Registry
  Viewer[Viewer] -->|hello, watch, sub| ViewerSession
  ViewerSession --> Registry
  Registry -->|subs snapshot| Robot
  Robot -->|data frames| Registry
  Registry -->|subscribed frames| Viewer

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart LR
  Robot[Robot bridge] -->|hello + manifest| RobotSession
  RobotSession --> Registry
  Viewer[Viewer] -->|hello, watch, sub| ViewerSession
  ViewerSession --> Registry
  Registry -->|subs snapshot| Robot
  Robot -->|data frames| Registry
  Registry -->|subscribed frames| Viewer

Loading

_{Reviews (1): Last reviewed commit: "fixes" | Re-trigger Greptile}

[greptile-apps]

Pre-handshake viewer routing

A client can send watch and then sub before sending any hello. These branches set watched and subs without checking viewer.greeted; onRobotFrame later routes frames using those fields without a greeted check. The client can therefore trigger a robot subscription snapshot and receive that channel's frames without completing the viewer handshake. The viewer path should require a valid role="viewer" hello before accepting watch/sub/unsub or forwarding frames.

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
2791	2	2789	165

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

dimos.web.relay_bridge.test_relay_bridge_module::test_encode_paths_and_max_hz_gate

Stack Traces | 0.098s run time

bridge = (<dimos.web.relay_bridge.relay_bridge_module.RelayBridgeModule object at 0xff8555be7a70>, [<dimos.web.relay_bridge.test_relay_bridge_module.FakeClient object at 0xff8556a03890>])

    def test_encode_paths_and_max_hz_gate(bridge) -> None:
        module, clients = bridge
        client = clients[0]
        push(module, client, Subs(chs=["color_image", "odom"], n=1))
        assert wait_until(
            lambda: image_transport(module).subscribers and odom_transport(module).subscribers
        )
    
        pose = PoseStamped(ts=42.5, position=[1.5, -2.5, 0.25], orientation=[0.0, 0.0, 0.0, 1.0])
        odom_transport(module).publish(pose)
        assert module.encoded["odom"] == 1
        assert wait_until(lambda: len(client.frames) == 1)
        ch, payload, delivery, _ = client.frames[0]
        assert (ch, delivery) == ("odom", "reliable")
        decoded = json.loads(payload)
        assert decoded == {"x": 1.5, "y": -2.5, "z": 0.25, "yaw": 0.0, "ts": 42.5}
    
        image = Image.from_numpy(np.zeros((8, 12, 3), dtype=np.uint8))
        image_transport(module).publish(image)
>       assert module.encoded["color_image"] == 1
E       assert 0 == 1

_          = None
bridge     = (<dimos.web.relay_bridge.relay_bridge_module.RelayBridgeModule object at 0xff8555be7a70>, [<dimos.web.relay_bridge.test_relay_bridge_module.FakeClient object at 0xff8556a03890>])
ch         = 'odom'
client     = <dimos.web.relay_bridge.test_relay_bridge_module.FakeClient object at 0xff8556a03890>
clients    = [<dimos.web.relay_bridge.test_relay_bridge_module.FakeClient object at 0xff8556a03890>]
decoded    = {'ts': 42.5, 'x': 1.5, 'y': -2.5, 'yaw': 0.0, ...}
delivery   = 'reliable'
image      = Image(shape=(8, 12, 3), format=BGR, dtype=uint8, frame_id='', ts=1784810997.825091)
module     = <dimos.web.relay_bridge.relay_bridge_module.RelayBridgeModule object at 0xff8555be7a70>
payload    = b'{"x":1.5,"y":-2.5,"z":0.25,"yaw":0.0,"ts":42.5}'
pose       = Pose(position=Vector([        1.5        -2.5        0.25]), orientation=Quaternion(0.000000, 0.000000, 0.000000, 1.000000))

.../web/relay_bridge/test_relay_bridge_module.py:289: AssertionError

dimos.web.relay_bridge.test_relay_bridge_e2e::test_full_session_flow_and_lazy_encode

Stack Traces | 16.6s run time

async def flow() -> None:
        assert bridge._url is not None
        async with await RelayClient.connect(bridge._url, "viewer") as viewer:
            # robots push carries the bridge's identity (drain until watch lands).
            await attach_viewer(viewer, ROBOT_ID, ["odom", "color_image"])
    
            frames = await collect_until(
                viewer,
                lambda fs: any(f.header.ch == "odom" for f in fs)
                and any(f.header.ch == "color_image" for f in fs),
                timeout=15.0,
            )
            odom = next(f for f in frames if f.header.ch == "odom")
            pose = json.loads(odom.payload)
            assert pose == {"x": 1.5, "y": -2.5, "z": 0.25, "yaw": 0.0, "ts": 42.5}
            assert odom.header.delivery == "reliable"
    
>           image = next(f for f in frames if f.header.ch == "color_image")
E           StopIteration

bridge     = <dimos.web.relay_bridge.relay_bridge_module.RelayBridgeModule object at 0xff435c9a9f70>
frames     = [DataFrame(header=FrameHeader(ch='odom', seq=0, ts=1784811102.9865143, delivery='reliable', meta=None), payload=b'{"x"...=1784811103.2389343, delivery='reliable', meta=None), payload=b'{"x":1.5,"y":-2.5,"z":0.25,"yaw":0.0,"ts":42.5}'), ...]
odom       = DataFrame(header=FrameHeader(ch='odom', seq=0, ts=1784811102.9865143, delivery='reliable', meta=None), payload=b'{"x":1.5,"y":-2.5,"z":0.25,"yaw":0.0,"ts":42.5}')
pose       = {'ts': 42.5, 'x': 1.5, 'y': -2.5, 'yaw': 0.0, ...}
viewer     = <dimos.web.relay_bridge.wt_client.RelayClient object at 0xff435c9d1ca0>

.../web/relay_bridge/test_relay_bridge_e2e.py:127: StopIteration

The above exception was the direct cause of the following exception:

bridge = <dimos.web.relay_bridge.relay_bridge_module.RelayBridgeModule object at 0xff435c9a9f70>
publisher = <dimos.web.relay_bridge.test_relay_bridge_e2e._Publisher object at 0xff435c98fe90>

    def test_full_session_flow_and_lazy_encode(
        bridge: RelayBridgeModule, publisher: _Publisher
    ) -> None:
        async def flow() -> None:
            assert bridge._url is not None
            async with await RelayClient.connect(bridge._url, "viewer") as viewer:
                # robots push carries the bridge's identity (drain until watch lands).
                await attach_viewer(viewer, ROBOT_ID, ["odom", "color_image"])
    
                frames = await collect_until(
                    viewer,
                    lambda fs: any(f.header.ch == "odom" for f in fs)
                    and any(f.header.ch == "color_image" for f in fs),
                    timeout=15.0,
                )
                odom = next(f for f in frames if f.header.ch == "odom")
                pose = json.loads(odom.payload)
                assert pose == {"x": 1.5, "y": -2.5, "z": 0.25, "yaw": 0.0, "ts": 42.5}
                assert odom.header.delivery == "reliable"
    
                image = next(f for f in frames if f.header.ch == "color_image")
                assert bytes(image.payload[:2]) == b"\xff\xd8"  # real TurboJPEG output
                assert image.header.meta == {"w": 64, "h": 48}
                assert image.header.delivery == "latest"
    
                # Unsub color_image: the bridge must stop encoding it entirely
                # while odom (still subscribed) keeps flowing.
                viewer.send_control(Unsub(ch="color_image"))
                deadline = time.monotonic() + 10
                while "color_image" in bridge._unsubs and time.monotonic() < deadline:
                    await asyncio.sleep(0.05)
                assert "color_image" not in bridge._unsubs, "bridge never heard the unsub"
    
                encoded_before = bridge.encoded["color_image"]
                await asyncio.sleep(0.5)  # publisher keeps publishing the whole time
                assert bridge.encoded["color_image"] == encoded_before
                assert "odom" in bridge._unsubs
    
>       asyncio.run(flow())

bridge     = <dimos.web.relay_bridge.relay_bridge_module.RelayBridgeModule object at 0xff435c9a9f70>
flow       = <function test_full_session_flow_and_lazy_encode.<locals>.flow at 0xff435b999580>
publisher  = <dimos.web.relay_bridge.test_relay_bridge_e2e._Publisher object at 0xff435c98fe90>

.../web/relay_bridge/test_relay_bridge_e2e.py:145: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
........./usr/lib/python3.12/asyncio/runners.py:194: in run
    return runner.run(main)
        debug      = None
        loop_factory = None
        main       = <coroutine object test_full_session_flow_and_lazy_encode.<locals>.flow at 0xff44193de440>
        runner     = <asyncio.runners.Runner object at 0xff435c972f90>
........./usr/lib/python3.12/asyncio/runners.py:118: in run
    return self._loop.run_until_complete(task)
        context    = <_contextvars.Context object at 0xff4445b18680>
        coro       = <coroutine object test_full_session_flow_and_lazy_encode.<locals>.flow at 0xff44193de440>
        self       = <asyncio.runners.Runner object at 0xff435c972f90>
        sigint_handler = functools.partial(<bound method Runner._on_sigint of <asyncio.runners.Runner object at 0xff435c972f90>>, main_task=<Ta...s/dimos/.../web/relay_bridge/test_relay_bridge_e2e.py:110> exception=RuntimeError('coroutine raised StopIteration')>)
        task       = <Task finished name='Task-218' coro=<test_full_session_flow_and_lazy_encode.<locals>.flow() done, defined at /home/run...os/dimos/.../web/relay_bridge/test_relay_bridge_e2e.py:110> exception=RuntimeError('coroutine raised StopIteration')>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_UnixSelectorEventLoop running=False closed=True debug=False>
future = <Task finished name='Task-218' coro=<test_full_session_flow_and_lazy_encode.<locals>.flow() done, defined at /home/run...os/dimos/.../web/relay_bridge/test_relay_bridge_e2e.py:110> exception=RuntimeError('coroutine raised StopIteration')>

    def run_until_complete(self, future):
        """Run until the Future is done.
    
        If the argument is a coroutine, it is wrapped in a Task.
    
        WARNING: It would be disastrous to call run_until_complete()
        with the same coroutine twice -- it would wrap it in two
        different Tasks and that can't be good.
    
        Return the Future's result, or raise its exception.
        """
        self._check_closed()
        self._check_running()
    
        new_task = not futures.isfuture(future)
        future = tasks.ensure_future(future, loop=self)
        if new_task:
            # An exception is raised if the future didn't complete, so there
            # is no need to log the "destroy pending task" message
            future._log_destroy_pending = False
    
        future.add_done_callback(_run_until_complete_cb)
        try:
            self.run_forever()
        except:
            if new_task and future.done() and not future.cancelled():
                # The coroutine raised a BaseException. Consume the exception
                # to not log a warning, the caller doesn't have access to the
                # local task.
                future.exception()
            raise
        finally:
            future.remove_done_callback(_run_until_complete_cb)
        if not future.done():
            raise RuntimeError('Event loop stopped before Future completed.')
    
>       return future.result()
E       RuntimeError: coroutine raised StopIteration

future     = <Task finished name='Task-218' coro=<test_full_session_flow_and_lazy_encode.<locals>.flow() done, defined at /home/run...os/dimos/.../web/relay_bridge/test_relay_bridge_e2e.py:110> exception=RuntimeError('coroutine raised StopIteration')>
new_task   = False
self       = <_UnixSelectorEventLoop running=False closed=True debug=False>

........./usr/lib/python3.12/asyncio/base_events.py:687: RuntimeError

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