feat(cockpit): map click-to-goal, path overlay and cancel button

dimos/e2e_tests/test_map_click_browser.py

@@ -0,0 +1,140 @@
+# Copyright 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.
+
+"""Browser-to-bridge coverage for map clicks and path-driven cancellation."""
+
+from collections.abc import Iterator
+import threading
+from typing import NamedTuple
+
+import pytest
+from reactivex.disposable import Disposable
+
+from dimos.core.transport import pLCMTransport
+from dimos.msgs.geometry_msgs.PointStamped import PointStamped
+from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
+from dimos.msgs.nav_msgs.Path import Path
+from dimos.web.cockpit import Map2D, cockpit
+from dimos.web.relay_bridge.e2e_support import stop_module
+from dimos.web.relay_bridge.gen_costmap_fixtures import grid_msg
+from dimos.web.relay_bridge.module_test_support import wait_until
+from dimos.web.relay_bridge.relay_bridge_module import RelayBridgeModule
+
+pytest.importorskip("playwright")
+
+from playwright.sync_api import Page, expect, sync_playwright
+
+pytestmark = pytest.mark.web_browser
+
+CANVAS = '[data-testid="map2d-global_costmap-canvas"]'
+# 2x2 cells at 0.5 m with the origin corner at (0.25, -0.5): world centre (0.75, 0).
+GRID = grid_msg([[0, 50], [100, -1]], 0.5, 0.25, -0.5, 0.0)
+
+
+class MapBridge(NamedTuple):
+    url: str
+    module: RelayBridgeModule
+    path: pLCMTransport
+
+
+def _path(*xy: tuple[float, float]) -> Path:
+    poses = [
+        PoseStamped(ts=1.0, position=[x, y, 0.0], orientation=[0.0, 0.0, 0.0, 1.0]) for x, y in xy
+    ]
+    return Path(ts=1.0, frame_id="world", poses=poses)
+
+
+@pytest.fixture(scope="module")
+def map_bridge() -> Iterator[MapBridge]:
+    layout = Map2D(pose=None, path="path", click="clicked_point", stop="stop_movement")
+    (atom,) = cockpit(layout=layout).blueprints
+    module = atom.module(local_port=0, open_browser=False, robot_id="map-click-e2e", **atom.kwargs)
+    transports: list[pLCMTransport] = []
+    publishers: dict[str, pLCMTransport] = {}
+    for ch in ("global_costmap", "path"):
+        topic = f"/map_click_e2e/{ch}"
+        bridge_side = pLCMTransport(topic)
+        bridge_side.start()
+        getattr(module, ch).transport = bridge_side
+        publishers[ch] = pLCMTransport(topic)
+        publishers[ch].start()
+        transports += [bridge_side, publishers[ch]]
+    stop = threading.Event()
+
+    def mapper() -> None:
+        # Frames flow only once the bridge lazily subscribes, so keep them coming.
+        while not stop.is_set():
+            publishers["global_costmap"].publish(GRID)
+            stop.wait(0.5)
+
+    thread = threading.Thread(target=mapper, daemon=True)
+    thread.start()
+    try:
+        module.start()
+        relay = module._relay
+        assert relay is not None and relay.info is not None
+        yield MapBridge(relay.info.open_url, module, publishers["path"])
+    finally:
+        stop.set()
+        thread.join(timeout=2)
+        stop_module(module)
+        for transport in transports:
+            transport.stop()
+
+
+@pytest.fixture
+def chromium_page() -> Iterator[Page]:
+    with sync_playwright() as p:
+        browser = p.chromium.launch()
+        try:
+            yield browser.new_page()
+        finally:
+            browser.close()
+
+
+def test_click_to_goal_and_cancel(map_bridge: MapBridge, chromium_page: Page) -> None:
+    clicks: list[PointStamped] = []
+    stops: list[bool] = []
+    map_bridge.module.register_disposable(
+        Disposable(map_bridge.module.clicked_point.subscribe(clicks.append))
+    )
+    map_bridge.module.register_disposable(
+        Disposable(map_bridge.module.stop_movement.subscribe(lambda msg: stops.append(msg.data)))
+    )
+
+    chromium_page.goto(map_bridge.url)
+    # The first grid sizes the backing store from the layout (300 is the
+    # unsized default): the map is drawn once that happens.
+    chromium_page.wait_for_function(
+        f"""() => {{
+          const canvas = document.querySelector('{CANVAS}');
+          return canvas !== null && canvas.width !== 300;
+        }}""",
+        timeout=120_000,
+    )
+    chromium_page.click(CANVAS)
+    assert wait_until(lambda: len(clicks) == 1, timeout=15.0)
+    assert clicks[0].x == pytest.approx(0.75, abs=0.01)
+    assert clicks[0].y == pytest.approx(0.0, abs=0.01)
+    assert clicks[0].frame_id == "world"
+
+    cancel = chromium_page.get_by_test_id("map2d-global_costmap-cancel")
+    expect(cancel).to_have_count(0)
+    map_bridge.path.publish(_path((0.3, -0.4), (0.75, 0.0), (1.2, 0.4)))
+    expect(cancel).to_be_visible(timeout=30_000)
+    cancel.click()
+    assert wait_until(lambda: stops == [True], timeout=15.0)
+
+    map_bridge.path.publish(Path())
+    expect(cancel).to_have_count(0, timeout=30_000)

dimos/navigation/basic_path_follower/module.py

@@ -19,7 +19,7 @@
 import time
 from typing import Any
 
-from dimos_lcm.std_msgs import Bool  # type: ignore[import-untyped]
+from dimos_lcm.std_msgs import Bool
 import numpy as np
 from numpy.typing import NDArray
 from reactivex.disposable import Disposable

dimos/navigation/dannav/holonomic_tc/module.py

@@ -31,7 +31,7 @@
 import traceback
 from typing import Any, Literal, TypeAlias
 
-from dimos_lcm.std_msgs import Bool  # type: ignore[import-untyped]
+from dimos_lcm.std_msgs import Bool
 import numpy as np
 from reactivex import Subject
 from reactivex.disposable import Disposable

dimos/navigation/dannav/holonomic_tc/test_dan_holonomic_tc.py

@@ -23,7 +23,7 @@
 import time
 from typing import Any, Literal
 
-from dimos_lcm.std_msgs import Bool  # type: ignore[import-untyped]
+from dimos_lcm.std_msgs import Bool
 import pytest
 
 from dimos.core.stream import Stream, Transport

dimos/navigation/movement_manager/movement_manager.py

@@ -25,7 +25,7 @@
 import time
 from typing import Any
 
-from dimos_lcm.std_msgs import Bool  # type: ignore[import-untyped]
+from dimos_lcm.std_msgs import Bool
 from reactivex.disposable import Disposable
 
 from dimos.core.core import rpc

dimos/robot/galaxea/r1pro/connection.py

@@ -111,7 +111,7 @@ def _make_qos() -> Any:
     """BEST_EFFORT + VOLATILE QoS — the profile the R1 Pro topics expect."""
     from rclpy.qos import DurabilityPolicy, QoSProfile, ReliabilityPolicy
 
-    return QoSProfile(
+    return QoSProfile(  # type: ignore[no-untyped-call]
         depth=10,
         reliability=ReliabilityPolicy.BEST_EFFORT,
         durability=DurabilityPolicy.VOLATILE,

dimos/robot/unitree/go2/blueprints/agentic/unitree_go2_agentic_cockpit.py

@@ -38,7 +38,7 @@
             layout=Row(
                 Video("color_image", title="Front camera"),
                 Col(
-                    Map2D(costmap="global_costmap", pose="odom", title="Map"),
+                    Map2D(path="path", click="clicked_point", stop="stop_movement", title="Map"),
                     Teleop(title="Keyboard teleop"),
                     shares=[3, 1],
                 ),

dimos/robot/unitree/go2/blueprints/smart/unitree_go2_cockpit.py

@@ -30,7 +30,11 @@
     cockpit(
         layout=Row(
             Video("color_image"),
-            Col(Map2D(costmap="global_costmap", pose="odom"), Teleop(), shares=[3, 1]),
+            Col(
+                Map2D(path="path", click="clicked_point", stop="stop_movement"),
+                Teleop(),
+                shares=[3, 1],
+            ),
             shares=[2, 1],
         ),
     ),

dimos/teleop/hosted/arm_command.py

@@ -38,6 +38,7 @@
 from dimos.teleop.webxr.controller_types import Hand
 from dimos.teleop.webxr.extensions import ArmTeleopModule
 from dimos.teleop.webxr.module import WebXRTeleopConfig
+from dimos.utils.generic import finite_number
 from dimos.utils.logging_config import setup_logger
 
 logger = setup_logger()
@@ -219,8 +220,8 @@ def _handle_teleop_scale(self, msg: dict[str, Any]) -> None:
             self._send_ack(nonce, False)
             return
         try:
-            self._set_translation_scale(float(msg["scale"]))
-        except (KeyError, TypeError, ValueError):
+            self._set_translation_scale(finite_number(msg.get("scale"), "scale"))
+        except ValueError:
             self._send_ack(nonce, False)
             return
         self._send_ack(nonce, True)

dimos/teleop/hosted/go2_command.py

@@ -39,6 +39,7 @@
 from dimos.msgs.geometry_msgs.TwistStamped import TwistStamped
 from dimos.robot.unitree.go2.connection import GO2Connection
 from dimos.teleop.hosted.command_executor import SerializedCommandExecutor
+from dimos.utils.generic import finite_number
 from dimos.utils.logging_config import setup_logger
 
 logger = setup_logger()
@@ -319,14 +320,11 @@ def _handle_light(self, msg: dict[str, Any]) -> None:
         if raw is None:
             raw = 1.0 if msg.get("enabled") else 0.0  # legacy on/off toggle
         try:
-            brightness = float(raw)
-        except (TypeError, ValueError):
+            brightness = finite_number(raw, "brightness")
+        except ValueError:
             logger.warning("light: malformed brightness %r", raw)
             self._send_ack(nonce, False)
             return
-        if math.isnan(brightness):
-            self._send_ack(nonce, False)
-            return
         brightness = max(0.0, min(1.0, brightness))
         level = round(brightness * 10)
 
@@ -349,13 +347,13 @@ def _handle_nav_goal(self, msg: dict[str, Any]) -> None:
             self._send_ack(nonce, False)
             return
         try:
-            x, y = float(msg["x"]), float(msg["y"])
-        except (KeyError, TypeError, ValueError):
+            x, y = finite_number(msg.get("x"), "x"), finite_number(msg.get("y"), "y")
+        except ValueError:
             logger.warning("nav_goal: malformed %r", msg)
             self._send_ack(nonce, False)
             return
         limit = self.config.max_nav_goal_m
-        if not (math.isfinite(x) and math.isfinite(y)) or abs(x) > limit or abs(y) > limit:
+        if abs(x) > limit or abs(y) > limit:
             logger.warning("nav_goal: out-of-range (%r, %r)", x, y)
             self._send_ack(nonce, False)
             return

dimos/teleop/hosted/test_go2_command.py

@@ -381,6 +381,18 @@ def test_nav_goal_publishes_and_acks(
     assert acks == [(11, True)]
 
 
+def test_nav_goal_rejects_non_numbers(
+    module: Go2CommandModule, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    acks: list[tuple[Any, bool]] = []
+    monkeypatch.setattr(module, "_send_ack", lambda nonce, ok: acks.append((nonce, ok)))
+
+    module._handle_nav_goal({"x": "2.5", "y": 1.0, "nonce": 17})
+
+    module.goal_request.publish.assert_not_called()
+    assert acks == [(17, False)]
+
+
 def test_nav_goal_rejected_when_estopped(
     module: Go2CommandModule, monkeypatch: pytest.MonkeyPatch
 ) -> None:

dimos/utils/generic.py

@@ -15,6 +15,7 @@
 from collections.abc import Callable
 import hashlib
 import json
+import math
 import os
 import socket
 import string
@@ -73,6 +74,13 @@ def extract_json_from_llm_response(response: str) -> Any:
     return None
 
 
+def finite_number(value: Any, name: str) -> float:
+    """Validate an untrusted JSON number: not a bool, not a string, finite."""
+    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
+        raise ValueError(f"{name} must be a finite number, got {value!r}")
+    return float(value)
+
+
 def short_id(from_string: str | None = None) -> str:
     alphabet = string.digits + string.ascii_letters
     base = len(alphabet)

dimos/utils/test_generic.py

@@ -12,9 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import math
 from uuid import UUID
 
-from dimos.utils.generic import short_id
+import pytest
+
+from dimos.utils.generic import finite_number, short_id
 
 
 def test_short_id_hello_world() -> None:
@@ -29,3 +32,14 @@ def test_short_id_uuid_one(mocker) -> None:
 def test_short_id_uuid_zero(mocker) -> None:
     mocker.patch("uuid.uuid4", return_value=UUID("00000000-0000-0000-0000-000000000000"))
     assert short_id() == "000000000000000000"
+
+
+def test_finite_number_coerces_to_float() -> None:
+    assert finite_number(2, "x") == 2.0
+    assert isinstance(finite_number(2, "x"), float)
+
+
+@pytest.mark.parametrize("value", [None, "1.5", True, math.nan, math.inf, [1.0]])
+def test_finite_number_rejects_non_numbers(value: object) -> None:
+    with pytest.raises(ValueError, match="x must be a finite number"):
+        finite_number(value, "x")