Jeff/fix/rconnect2

dimos/core/coordination/python_worker.py

@@ -18,6 +18,7 @@
 import multiprocessing
 from multiprocessing.connection import Connection
 import os
+import signal
 import sys
 import threading
 import traceback
@@ -337,12 +338,15 @@ class _WorkerState:
 
 def _worker_entrypoint(conn: Connection, worker_id: int) -> None:
     apply_library_config()
+    # Ignore SIGINT so the coordinator can orchestrate shutdown via the pipe.
+    # Without this, workers race with the coordinator: they start tearing down
+    # modules locally while the coordinator tries to send stop() RPCs, causing
+    # BrokenPipeErrors.
+    signal.signal(signal.SIGINT, signal.SIG_IGN)
     state = _WorkerState(instances={}, worker_id=worker_id)
 
     try:
         _worker_loop(conn, state)
-    except KeyboardInterrupt:
-        logger.info("Worker got KeyboardInterrupt.", worker_id=worker_id)
     except Exception as e:
         logger.error(f"Worker process error: {e}", exc_info=True)
     finally:
@@ -361,12 +365,6 @@ def _worker_entrypoint(conn: Connection, worker_id: int) -> None:
                     worker_id=worker_id,
                     module_id=module_id,
                 )
-            except KeyboardInterrupt:
-                logger.warning(
-                    "KeyboardInterrupt during worker stop",
-                    module=type(instance).__name__,
-                    worker_id=worker_id,
-                )
             except Exception:
                 logger.error("Error during worker shutdown", exc_info=True)
 
@@ -433,7 +431,7 @@ def _worker_loop(conn: Connection, state: _WorkerState) -> None:
             if not conn.poll(timeout=0.1):
                 continue
             request = conn.recv()
-        except (EOFError, KeyboardInterrupt):
+        except EOFError:
             break
 
         try:

dimos/core/docker_module.py

@@ -30,7 +30,7 @@
 from dimos.core.rpc_client import ModuleProxyProtocol, RpcCall
 from dimos.protocol.rpc.pubsubrpc import LCMRPC
 from dimos.utils.logging_config import setup_logger
-from dimos.visualization.rerun.bridge import RERUN_GRPC_PORT, RERUN_WEB_PORT
+from dimos.visualization.rerun.constants import RERUN_GRPC_PORT, RERUN_WEB_PORT
 
 if TYPE_CHECKING:
     from collections.abc import Callable

dimos/core/global_config.py

@@ -13,13 +13,16 @@
 # limitations under the License.
 
 import re
-from typing import Literal, TypeAlias
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 from dimos.models.vl.types import VlModelName
-
-ViewerBackend: TypeAlias = Literal["rerun", "rerun-web", "rerun-connect", "foxglove", "none"]
+from dimos.visualization.rerun.constants import (
+    RERUN_ENABLE_WEB,
+    RERUN_OPEN_DEFAULT,
+    RerunOpenOption,
+    ViewerBackend,
+)
 
 
 def _get_all_numbers(s: str) -> list[float]:
@@ -37,6 +40,8 @@ class GlobalConfig(BaseSettings):
     replay_dir: str = "go2_sf_office"
     new_memory: bool = False
     viewer: ViewerBackend = "rerun"
+    rerun_open: RerunOpenOption = RERUN_OPEN_DEFAULT
+    rerun_web: bool = RERUN_ENABLE_WEB
     n_workers: int = 2
     memory_limit: str = "auto"
     mujoco_camera_position: str | None = None

dimos/hardware/sensors/camera/module.py

@@ -21,6 +21,7 @@
 from dimos.agents.annotation import skill
 from dimos.core.coordination.blueprints import autoconnect
 from dimos.core.core import rpc
+from dimos.core.global_config import global_config
 from dimos.core.module import Module, ModuleConfig
 from dimos.core.stream import Out
 from dimos.hardware.sensors.camera.spec import CameraHardware
@@ -31,7 +32,7 @@
 from dimos.msgs.sensor_msgs.CameraInfo import CameraInfo
 from dimos.msgs.sensor_msgs.Image import Image, sharpness_barrier
 from dimos.spec import perception
-from dimos.visualization.rerun.bridge import RerunBridgeModule
+from dimos.visualization.vis_module import vis_module
 
 
 def default_transform() -> Transform:
@@ -120,5 +121,5 @@ def stop(self) -> None:
 
 demo_camera = autoconnect(
     CameraModule.blueprint(),
-    RerunBridgeModule.blueprint(),
+    vis_module(viewer_backend=global_config.viewer),
 )

dimos/hardware/sensors/lidar/fastlio2/fastlio_blueprints.py

@@ -15,30 +15,45 @@
 from dimos.core.coordination.blueprints import autoconnect
 from dimos.hardware.sensors.lidar.fastlio2.module import FastLio2
 from dimos.mapping.voxels import VoxelGridMapper
-from dimos.visualization.rerun.bridge import RerunBridgeModule
+from dimos.visualization.vis_module import vis_module
 
 voxel_size = 0.05
 
 mid360_fastlio = autoconnect(
     FastLio2.blueprint(voxel_size=voxel_size, map_voxel_size=voxel_size, map_freq=-1),
-    RerunBridgeModule.blueprint(),
+    vis_module(
+        "rerun",
+        rerun_config={
+            "visual_override": {
+                "world/lidar": lambda grid: grid.to_rerun(voxel_size=voxel_size, mode="boxes"),
+            },
+        },
+    ),
 ).global_config(n_workers=2, robot_model="mid360_fastlio2")
 
 mid360_fastlio_voxels = autoconnect(
     FastLio2.blueprint(),
     VoxelGridMapper.blueprint(voxel_size=voxel_size, carve_columns=False),
-    RerunBridgeModule.blueprint(
-        visual_override={
-            "world/lidar": None,
-        }
+    vis_module(
+        "rerun",
+        rerun_config={
+            "visual_override": {
+                "world/global_map": lambda grid: grid.to_rerun(voxel_size=voxel_size, mode="boxes"),
+                "world/lidar": None,
+            },
+        },
     ),
 ).global_config(n_workers=3, robot_model="mid360_fastlio2_voxels")
 
 mid360_fastlio_voxels_native = autoconnect(
     FastLio2.blueprint(voxel_size=voxel_size, map_voxel_size=voxel_size, map_freq=3.0),
-    RerunBridgeModule.blueprint(
-        visual_override={
-            "world/lidar": None,
-        }
+    vis_module(
+        "rerun",
+        rerun_config={
+            "visual_override": {
+                "world/lidar": None,
+                "world/global_map": lambda grid: grid.to_rerun(voxel_size=voxel_size, mode="boxes"),
+            },
+        },
     ),
 ).global_config(n_workers=2, robot_model="mid360_fastlio2")

dimos/hardware/sensors/lidar/livox/livox_blueprints.py

@@ -14,9 +14,9 @@
 
 from dimos.core.coordination.blueprints import autoconnect
 from dimos.hardware.sensors.lidar.livox.module import Mid360
-from dimos.visualization.rerun.bridge import RerunBridgeModule
+from dimos.visualization.vis_module import vis_module
 
 mid360 = autoconnect(
     Mid360.blueprint(),
-    RerunBridgeModule.blueprint(),
+    vis_module("rerun"),
 ).global_config(n_workers=2, robot_model="mid360")

dimos/manipulation/blueprints.py

@@ -44,7 +44,7 @@
 from dimos.msgs.sensor_msgs.JointState import JointState
 from dimos.perception.object_scene_registration import ObjectSceneRegistrationModule
 from dimos.robot.catalog.ufactory import xarm6 as _catalog_xarm6, xarm7 as _catalog_xarm7
-from dimos.robot.foxglove_bridge import FoxgloveBridge  # TODO: migrate to rerun
+from dimos.visualization.vis_module import vis_module
 
 # Single XArm6 planner (standalone, no coordinator)
 _xarm6_planner_cfg = _catalog_xarm6(
@@ -196,14 +196,14 @@
             use_aabb=True,
             max_obstacle_width=0.06,
         ),
-        FoxgloveBridge.blueprint(),  # TODO: migrate to rerun
+        vis_module("foxglove"),
     )
     .transports(
         {
             ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
         }
     )
-    .global_config(viewer="foxglove", n_workers=4)
+    .global_config(n_workers=4)
 )
 
 
@@ -289,7 +289,7 @@
 
 from dimos.robot.catalog.ufactory import XARM7_SIM_PATH
 from dimos.simulation.engines.mujoco_sim_module import MujocoSimModule
-from dimos.visualization.rerun.bridge import RerunBridgeModule, _resolve_viewer_mode
+from dimos.visualization.rerun.bridge import RerunBridgeModule
 
 _xarm7_sim_cfg = _catalog_xarm7(
     name="arm",
@@ -323,7 +323,7 @@
         hardware=[_xarm7_sim_cfg.to_hardware_component()],
         tasks=[_xarm7_sim_cfg.to_task_config()],
     ),
-    RerunBridgeModule.blueprint(viewer_mode=_resolve_viewer_mode()),
+    RerunBridgeModule.blueprint(),
 ).transports(
     {
         ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),

dimos/manipulation/grasping/demo_grasping.py

@@ -22,7 +22,7 @@
 from dimos.manipulation.grasping.grasping import GraspingModule
 from dimos.perception.detection.detectors.yoloe import YoloePromptMode
 from dimos.perception.object_scene_registration import ObjectSceneRegistrationModule
-from dimos.robot.foxglove_bridge import FoxgloveBridge
+from dimos.visualization.vis_module import vis_module
 
 camera_module = RealSenseCamera.blueprint(enable_pointcloud=False)
 
@@ -44,7 +44,7 @@
             ("/tmp", "/tmp", "rw")
         ],  # Grasp visualization debug standalone: python -m dimos.manipulation.grasping.visualize_grasps
     ),
-    FoxgloveBridge.blueprint(),
+    vis_module("foxglove"),
     McpServer.blueprint(),
     McpClient.blueprint(),
 ).global_config(viewer="foxglove")

dimos/navigation/frontier_exploration/wavefront_frontier_goal_selector.py

@@ -115,6 +115,7 @@ class WavefrontFrontierExplorer(Module):
     goal_reached: In[Bool]
     explore_cmd: In[Bool]
     stop_explore_cmd: In[Bool]
+    stop_movement: In[Bool]
 
     # LCM outputs
     goal_request: Out[PoseStamped]
@@ -171,6 +172,10 @@ def start(self) -> None:
             unsub = self.stop_explore_cmd.subscribe(self._on_stop_explore_cmd)
             self.register_disposable(Disposable(unsub))
 
+        if self.stop_movement.transport is not None:
+            unsub = self.stop_movement.subscribe(self._on_stop_movement)
+            self.register_disposable(Disposable(unsub))
+
     @rpc
     def stop(self) -> None:
         self.stop_exploration()
@@ -201,6 +206,12 @@ def _on_stop_explore_cmd(self, msg: Bool) -> None:
             logger.info("Received exploration stop command via LCM")
             self.stop_exploration()
 
+    def _on_stop_movement(self, msg: Bool) -> None:
+        """Handle stop movement from teleop — cancel active exploration."""
+        if msg.data and self.exploration_active:
+            logger.info("WavefrontFrontierExplorer: stop_movement received, stopping exploration")
+            self.stop_exploration()
+
     def _count_costmap_information(self, costmap: OccupancyGrid) -> int:
         """
         Count the amount of information in a costmap (free space + obstacles).

dimos/navigation/replanning_a_star/module.py

@@ -28,6 +28,9 @@
 from dimos.msgs.nav_msgs.Path import Path
 from dimos.navigation.base import NavigationInterface, NavigationState
 from dimos.navigation.replanning_a_star.global_planner import GlobalPlanner
+from dimos.utils.logging_config import setup_logger
+
+logger = setup_logger()
 
 
 class ReplanningAStarPlanner(Module, NavigationInterface):
@@ -36,10 +39,11 @@ class ReplanningAStarPlanner(Module, NavigationInterface):
     goal_request: In[PoseStamped]
     clicked_point: In[PointStamped]
     target: In[PoseStamped]
+    stop_movement: In[Bool]
 
     goal_reached: Out[Bool]
     navigation_state: Out[String]  # TODO: set it
-    cmd_vel: Out[Twist]
+    nav_cmd_vel: Out[Twist]
     path: Out[Path]
     navigation_costmap: Out[OccupancyGrid]
 
@@ -72,9 +76,14 @@ def start(self) -> None:
             )
         )
 
+        if self.stop_movement.transport is not None:
+            self.register_disposable(
+                Disposable(self.stop_movement.subscribe(self._on_stop_movement))
+            )
+
         self.register_disposable(self._planner.path.subscribe(self.path.publish))
 
-        self.register_disposable(self._planner.cmd_vel.subscribe(self.cmd_vel.publish))
+        self.register_disposable(self._planner.cmd_vel.subscribe(self.nav_cmd_vel.publish))
 
         self.register_disposable(self._planner.goal_reached.subscribe(self.goal_reached.publish))
 
@@ -92,6 +101,11 @@ def stop(self) -> None:
 
         super().stop()
 
+    def _on_stop_movement(self, msg: Bool) -> None:
+        if msg.data:
+            logger.info("ReplanningAStarPlanner: stop_movement received, cancelling goal")
+            self.cancel_goal()
+
     @rpc
     def set_goal(self, goal: PoseStamped) -> bool:
         self._planner.handle_goal_request(goal)