feat(dimsim): add closed-loop agent evaluations

.github/workflows/dimsim-check.yml

@@ -47,3 +47,6 @@ jobs:
 
       - name: Type-check CLI
         run: cd cli && deno check cli.ts
+
+      - name: Run Deno tests
+        run: deno test -A --unstable-net --config cli/deno.json cli evals

dimos/robot/unitree/dimsim_connection.py

@@ -16,7 +16,7 @@
 import functools
 from typing import Any
 
-from reactivex import Observable, Subject
+from reactivex import Subject
 
 from dimos.core.global_config import GlobalConfig
 from dimos.core.transport import PubSubTransport
@@ -51,36 +51,58 @@ class DimSimConnection:
     def __init__(self, global_config: GlobalConfig) -> None:
         self._dimsim_process: DimSimProcess = DimSimProcess(global_config)
         self._odom_transport: PubSubTransport[PoseStamped] = make_transport("/odom", PoseStamped)
-        self._unsubscribe_odom: Callable[[], None] | None = None
+        self._lidar_transport: PubSubTransport[PointCloud2] = make_transport("/lidar", PointCloud2)
+        self._video_transport: PubSubTransport[Image] = make_transport("/color_image", Image)
+        self._unsubscribes: list[Callable[[], None]] = []
+        self._latest_sensor_ts = {
+            "odom": float("-inf"),
+            "lidar": float("-inf"),
+            "video": float("-inf"),
+        }
         self._tf = tf_backend()()
 
     def start(self) -> None:
         self._dimsim_process.start()
-        self._odom_transport.start()
-        self._unsubscribe_odom = self._odom_transport.subscribe(self._handle_odom)
+        for transport in (
+            self._odom_transport,
+            self._lidar_transport,
+            self._video_transport,
+        ):
+            transport.start()
+        self._unsubscribes = [
+            self._odom_transport.subscribe(self._handle_odom),
+            self._lidar_transport.subscribe(self._handle_lidar),
+            self._video_transport.subscribe(self._handle_video),
+        ]
         self._tf.start()
 
     def stop(self) -> None:
         self._tf.stop()
-        if self._unsubscribe_odom is not None:
-            self._unsubscribe_odom()
-        self._odom_transport.stop()
+        for unsubscribe in self._unsubscribes:
+            unsubscribe()
+        self._unsubscribes.clear()
+        for transport in (
+            self._video_transport,
+            self._lidar_transport,
+            self._odom_transport,
+        ):
+            transport.stop()
         self._dimsim_process.stop()
 
     @functools.cache
-    def lidar_stream(self) -> Observable[PointCloud2]:
+    def lidar_stream(self) -> Subject[PointCloud2]:
         return Subject()
 
     @functools.cache
-    def odom_stream(self) -> Observable[PoseStamped]:
+    def odom_stream(self) -> Subject[PoseStamped]:
         return Subject()
 
     @functools.cache
-    def video_stream(self) -> Observable[Image]:
+    def video_stream(self) -> Subject[Image]:
         return Subject()
 
     @functools.cache
-    def lowstate_stream(self) -> Observable[Any]:
+    def lowstate_stream(self) -> Subject[Any]:
         return Subject()
 
     def move(self, twist: Twist, duration: float = 0.0) -> bool:
@@ -118,7 +140,25 @@ def publish_request(self, topic: str, data: dict[str, Any]) -> dict[Any, Any]:
         return {}
 
     def _handle_odom(self, msg: PoseStamped) -> None:
+        if not self._is_new_sensor_sample("odom", msg.ts):
+            return
         self._tf.publish(*_odom_to_tf(msg))
+        self.odom_stream().on_next(msg)
+
+    def _handle_lidar(self, msg: PointCloud2) -> None:
+        if self._is_new_sensor_sample("lidar", msg.ts):
+            self.lidar_stream().on_next(msg)
+
+    def _handle_video(self, msg: Image) -> None:
+        if self._is_new_sensor_sample("video", msg.ts):
+            self.video_stream().on_next(msg)
+
+    def _is_new_sensor_sample(self, stream: str, timestamp: float) -> bool:
+        """Reject the same packet when GO2 republishes it on the bridge topic."""
+        if timestamp <= self._latest_sensor_ts[stream]:
+            return False
+        self._latest_sensor_ts[stream] = timestamp
+        return True
 
 
 def _odom_to_tf(odom: PoseStamped) -> list[Transform]:

dimos/robot/unitree/test_dimsim_connection.py

@@ -0,0 +1,115 @@
+# Copyright 2025-2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from dimos.core.global_config import GlobalConfig
+from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
+from dimos.msgs.sensor_msgs.Image import Image
+from dimos.msgs.sensor_msgs.PointCloud2 import PointCloud2
+from dimos.robot.unitree.dimsim_connection import DimSimConnection
+
+
+def test_dimsim_camera_info_matches_browser_sensor_frames():
+    camera_info = DimSimConnection.camera_info_static
+
+    assert camera_info.width == 640
+    assert camera_info.height == 288
+
+
+@pytest.fixture
+def dimsim_connection(mocker):
+    process = mocker.patch("dimos.robot.unitree.dimsim_connection.DimSimProcess").return_value
+    transports = [MagicMock(), MagicMock(), MagicMock()]
+    mocker.patch(
+        "dimos.robot.unitree.dimsim_connection.make_transport",
+        side_effect=transports,
+    )
+    tf = MagicMock()
+    mocker.patch("dimos.robot.unitree.dimsim_connection.tf_backend").return_value.return_value = tf
+
+    callbacks = []
+    unsubscribes = []
+    for transport in transports:
+        unsubscribe = MagicMock()
+        unsubscribes.append(unsubscribe)
+
+        def subscribe(callback, *, _unsubscribe=unsubscribe):
+            callbacks.append(callback)
+            return _unsubscribe
+
+        transport.subscribe.side_effect = subscribe
+
+    connection = DimSimConnection(GlobalConfig(simulation="dimsim"))
+    connection.start()
+    try:
+        yield connection, process, transports, tf, callbacks, unsubscribes
+    finally:
+        connection.stop()
+
+
+def test_dimsim_connection_relays_sensor_topics(dimsim_connection):
+    connection, _, _, tf, callbacks, _ = dimsim_connection
+    received_odom = []
+    received_lidar = []
+    received_video = []
+    connection.odom_stream().subscribe(received_odom.append)
+    connection.lidar_stream().subscribe(received_lidar.append)
+    connection.video_stream().subscribe(received_video.append)
+
+    odom = PoseStamped(ts=1.0)
+    lidar = PointCloud2(ts=1.0)
+    image = Image(ts=1.0)
+    callbacks[0](odom)
+    callbacks[1](lidar)
+    callbacks[2](image)
+
+    assert received_odom == [odom]
+    assert received_lidar == [lidar]
+    assert received_video == [image]
+    assert tf.publish.call_count == 1
+
+
+def test_dimsim_connection_drops_republished_sensor_packets(dimsim_connection):
+    connection, _, _, _, callbacks, _ = dimsim_connection
+    received_odom = []
+    received_lidar = []
+    received_video = []
+    connection.odom_stream().subscribe(received_odom.append)
+    connection.lidar_stream().subscribe(received_lidar.append)
+    connection.video_stream().subscribe(received_video.append)
+
+    odom = PoseStamped(ts=1.0)
+    lidar = PointCloud2(ts=1.0)
+    image = Image(ts=1.0)
+    for callback, message in zip(callbacks, (odom, lidar, image), strict=True):
+        callback(message)
+        callback(message)
+
+    assert received_odom == [odom]
+    assert received_lidar == [lidar]
+    assert received_video == [image]
+
+
+def test_dimsim_connection_stops_all_resources(dimsim_connection):
+    connection, process, transports, tf, _, unsubscribes = dimsim_connection
+
+    connection.stop()
+
+    assert [unsubscribe.call_count for unsubscribe in unsubscribes] == [1, 1, 1]
+    assert [transport.stop.call_count for transport in transports] == [1, 1, 1]
+    tf.stop.assert_called_once_with()
+    process.stop.assert_called_once_with()