diff --git a/dimos/codebase_checks/test_get_logger.py b/dimos/codebase_checks/test_get_logger.py index 26db11ab8f..f2308cf9c0 100644 --- a/dimos/codebase_checks/test_get_logger.py +++ b/dimos/codebase_checks/test_get_logger.py @@ -48,6 +48,10 @@ "dimos/visualization/rerun/websocket_server.py", 'ws_logger = logging.getLogger("websockets.server")', ), + ( + "dimos/utils/cli/spy/conftest.py", + "lg = logging.getLogger(_CORE_LOGGER)", + ), ] diff --git a/dimos/robot/cli/dimos.py b/dimos/robot/cli/dimos.py index ba7a9b3015..136c85c554 100644 --- a/dimos/robot/cli/dimos.py +++ b/dimos/robot/cli/dimos.py @@ -692,13 +692,34 @@ def list_blueprints() -> None: typer.echo(blueprint_name) +@main.command(context_settings={"allow_extra_args": True, "ignore_unknown_options": True}) +def spy(ctx: typer.Context) -> None: + """Universal transport spy: topics, rates, sizes across all pubsub transports.""" + # A root-level `--transport` (before the subcommand) sets the stack's pubsub + # backend — which single transport dimos processes participate on. The spy is an + # observer: it watches every transport and takes its own repeatable `--transport` + # filter *after* the subcommand. The two look alike but mean different things, so + # reject the root placement rather than silently ignoring the requested filter. + if (ctx.obj or {}).get("transport") is not None: + typer.echo( + "Error: `--transport` before `spy` sets the stack backend, which the spy " + "ignores. Put the filter after the subcommand: `dimos spy --transport `.", + err=True, + ) + raise typer.Exit(2) + from dimos.utils.cli.spy.run_spy import main as spy_main + + sys.argv = ["spy", *ctx.args] + spy_main() + + @main.command(context_settings={"allow_extra_args": True, "ignore_unknown_options": True}) def lcmspy(ctx: typer.Context) -> None: - """LCM spy tool for monitoring LCM messages.""" - from dimos.utils.cli.lcmspy.run_lcmspy import main as lcmspy_main + """Alias for `dimos spy --transport lcm`.""" + from dimos.utils.cli.spy.run_spy import lcm_only_argv, main as spy_main - sys.argv = ["lcmspy", *ctx.args] - lcmspy_main() + sys.argv = lcm_only_argv(list(ctx.args)) + spy_main() @main.command(context_settings={"allow_extra_args": True, "ignore_unknown_options": True}) diff --git a/dimos/robot/cli/test_dimos.py b/dimos/robot/cli/test_dimos.py index cc3c26e04a..cd82d05c33 100644 --- a/dimos/robot/cli/test_dimos.py +++ b/dimos/robot/cli/test_dimos.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import sys from typing import Literal from pydantic import BaseModel, Field @@ -22,6 +23,7 @@ from dimos.core.global_config import global_config from dimos.core.module import Module, ModuleConfig from dimos.robot.cli.dimos import _normalize_simulation_argv, arg_help, main +import dimos.utils.cli.spy.run_spy as run_spy @pytest.mark.parametrize( @@ -163,6 +165,58 @@ class TestModule(Module): ] +@pytest.fixture +def spy_main_argv(monkeypatch): + """Stub run_spy.main and capture the sys.argv the lcmspy alias hands it.""" + captured: list[list[str]] = [] + monkeypatch.setattr(sys, "argv", ["dimos"]) + monkeypatch.setattr(run_spy, "main", lambda: captured.append(list(sys.argv))) + return captured + + +def test_lcmspy_alias_prepends_lcm_transport(spy_main_argv): + result = CliRunner().invoke(main, ["lcmspy"]) + assert result.exit_code == 0, result.output + assert spy_main_argv == [["spy", "--transport", "lcm"]] + + +def test_lcmspy_alias_rejects_transport_override(spy_main_argv): + result = CliRunner().invoke(main, ["lcmspy", "--transport", "zenoh"]) + assert result.exit_code == 1 + assert "LCM-only" in result.output + assert spy_main_argv == [] # never reaches the spy + + +def test_spy_cmd_rejects_stray_positional(monkeypatch): + # A stray positional must fail loudly, not silently launch the TUI. + monkeypatch.setattr(sys, "argv", ["dimos"]) + result = CliRunner().invoke(main, ["spy", "foo"]) + assert result.exit_code == 1 + assert "unexpected" in result.output.lower() + + +def test_lcmspy_rejects_stray_positional(monkeypatch): + monkeypatch.setattr(sys, "argv", ["dimos"]) + result = CliRunner().invoke(main, ["lcmspy", "foo"]) + assert result.exit_code == 1 + assert "unexpected" in result.output.lower() + + +def test_spy_rejects_root_transport(monkeypatch, spy_main_argv): + # A root-level `--transport` (before the subcommand) sets the stack backend, + # which the spy ignores. Rather than silently show all transports, error and + # point at the subcommand-level filter. + monkeypatch.setattr(sys, "argv", ["dimos"]) + original = global_config.transport + try: + result = CliRunner().invoke(main, ["--transport", "zenoh", "spy"]) + finally: + global_config.update(transport=original) + assert result.exit_code == 2 + assert "dimos spy --transport" in result.output + assert spy_main_argv == [] # never reaches the spy + + def test_blueprint_arg_help_nested_config_paths(): class NestedConfig(BaseModel): enabled: bool = True diff --git a/dimos/utils/cli/lcmspy/lcmspy.py b/dimos/utils/cli/lcmspy/lcmspy.py deleted file mode 100755 index b1fdc2db38..0000000000 --- a/dimos/utils/cli/lcmspy/lcmspy.py +++ /dev/null @@ -1,196 +0,0 @@ -# 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 collections import deque -import threading -import time -from typing import Any - -from dimos.constants import DEFAULT_THREAD_JOIN_TIMEOUT -from dimos.protocol.service.lcmservice import LCMConfig, LCMService -from dimos.utils.human import human_bytes - - -class Topic: - history_window: float = 60.0 - - def __init__(self, name: str, history_window: float = 60.0) -> None: - self.name = name - # Store (timestamp, data_size) tuples for statistics - self.message_history = deque() # type: ignore[var-annotated] - self._lock = threading.Lock() - self.history_window = history_window - # Total traffic accumulator (doesn't get cleaned up) - self.total_traffic_bytes = 0 - - def msg(self, data: bytes) -> None: - # print(f"> msg {self.__str__()} {len(data)} bytes") - datalen = len(data) - with self._lock: - self.message_history.append((time.time(), datalen)) - self.total_traffic_bytes += datalen - self._cleanup_old_messages() - - def _cleanup_old_messages(self, max_age: float | None = None) -> None: - """Remove messages older than max_age seconds""" - current_time = time.time() - while self.message_history and current_time - self.message_history[0][0] > ( - max_age or self.history_window - ): - self.message_history.popleft() - - def _get_messages_in_window(self, time_window: float): # type: ignore[no-untyped-def] - """Get messages within the specified time window""" - current_time = time.time() - cutoff_time = current_time - time_window - with self._lock: - return [(ts, size) for ts, size in self.message_history if ts >= cutoff_time] - - # avg msg freq in the last n seconds - def freq(self, time_window: float) -> float: - messages = self._get_messages_in_window(time_window) - if not messages: - return 0.0 - return len(messages) / time_window - - # avg bandwidth in kB/s in the last n seconds - def kbps(self, time_window: float) -> float: - messages = self._get_messages_in_window(time_window) - if not messages: - return 0.0 - total_bytes = sum(size for _, size in messages) - total_kbytes = total_bytes / 1000 # Convert bytes to kB - return total_kbytes / time_window # type: ignore[no-any-return] - - def kbps_hr(self, time_window: float) -> str: - """Return human-readable bandwidth with appropriate units""" - bps = self.kbps(time_window) * 1000 - return human_bytes(bps) + "/s" - - # avg msg size in the last n seconds - def size(self, time_window: float) -> float: - messages = self._get_messages_in_window(time_window) - if not messages: - return 0.0 - total_size = sum(size for _, size in messages) - return total_size / len(messages) # type: ignore[no-any-return] - - def total_traffic(self) -> int: - """Return total traffic passed in bytes since the beginning""" - with self._lock: - return self.total_traffic_bytes - - def total_traffic_hr(self) -> str: - """Return human-readable total traffic with appropriate units""" - return human_bytes(self.total_traffic()) - - def __str__(self) -> str: - return f"topic({self.name})" - - -class LCMSpyConfig(LCMConfig): - topic_history_window: float = 60.0 - - -class LCMSpy(LCMService, Topic): - config: LCMSpyConfig - topic = dict[str, Topic] - graph_log_window: float = 1.0 - topic_class: type[Topic] = Topic - - def __init__(self, **kwargs: Any) -> None: - super().__init__(**kwargs) - Topic.__init__(self, name="total", history_window=self.config.topic_history_window) - self.topic = {} # type: ignore[assignment] - self._topic_lock = threading.Lock() - - def start(self) -> None: - super().start() - self.l.subscribe(".*", self.msg) # type: ignore[union-attr] - - def stop(self) -> None: - """Stop the LCM spy and clean up resources""" - super().stop() - - def msg(self, topic, data) -> None: # type: ignore[no-untyped-def, override] - Topic.msg(self, data) - - with self._topic_lock: - if topic not in self.topic: # type: ignore[operator] - print(self.config) - self.topic[topic] = self.topic_class( # type: ignore[assignment, call-arg] - topic, - history_window=self.config.topic_history_window, - ) - self.topic[topic].msg(data) # type: ignore[attr-defined, type-arg] - - -class GraphTopic(Topic): - def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] - super().__init__(*args, **kwargs) - self.freq_history = deque(maxlen=20) # type: ignore[var-annotated] - self.bandwidth_history = deque(maxlen=20) # type: ignore[var-annotated] - - def update_graphs(self, step_window: float = 1.0) -> None: - """Update historical data for graphing""" - freq = self.freq(step_window) - kbps = self.kbps(step_window) - self.freq_history.append(freq) - self.bandwidth_history.append(kbps) - - -class GraphLCMSpyConfig(LCMSpyConfig): - graph_log_window: float = 1.0 - - -class GraphLCMSpy(LCMSpy, GraphTopic): - config: GraphLCMSpyConfig - graph_log_thread: threading.Thread | None = None - graph_log_stop_event: threading.Event = threading.Event() - topic_class: type[Topic] = GraphTopic - - def __init__(self, **kwargs: Any) -> None: - super().__init__(**kwargs) - GraphTopic.__init__(self, name="total", history_window=self.config.topic_history_window) - - def start(self) -> None: - super().start() - self.graph_log_thread = threading.Thread(target=self.graph_log, daemon=True) - self.graph_log_thread.start() - - def graph_log(self) -> None: - while not self.graph_log_stop_event.is_set(): - self.update_graphs(self.config.graph_log_window) # type: ignore[attr-defined] # Update global history - with self._topic_lock: - topics = list(self.topic.values()) # type: ignore[call-arg] - for topic in topics: - topic.update_graphs(self.config.graph_log_window) # type: ignore[attr-defined] - time.sleep(self.config.graph_log_window) # type: ignore[attr-defined] - - def stop(self) -> None: - """Stop the graph logging and LCM spy""" - self.graph_log_stop_event.set() - if self.graph_log_thread and self.graph_log_thread.is_alive(): - self.graph_log_thread.join(timeout=DEFAULT_THREAD_JOIN_TIMEOUT) - super().stop() - - -if __name__ == "__main__": - lcm_spy = LCMSpy() - lcm_spy.start() - try: - while True: - time.sleep(1) - except KeyboardInterrupt: - print("LCM Spy stopped.") diff --git a/dimos/utils/cli/lcmspy/run_lcmspy.py b/dimos/utils/cli/lcmspy/run_lcmspy.py deleted file mode 100644 index 438d93fa1f..0000000000 --- a/dimos/utils/cli/lcmspy/run_lcmspy.py +++ /dev/null @@ -1,138 +0,0 @@ -# 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 __future__ import annotations - -from rich.text import Text -from textual.app import App, ComposeResult -from textual.color import Color -from textual.widgets import DataTable - -from dimos.utils.cli import theme -from dimos.utils.cli.lcmspy.lcmspy import GraphLCMSpy, GraphTopic as SpyTopic - - -def gradient(max_value: float, value: float) -> str: - """Gradient from cyan (low) to yellow (high) using DimOS theme colors""" - ratio = min(value / max_value, 1.0) - # Parse hex colors from theme - cyan = Color.parse(theme.CYAN) - yellow = Color.parse(theme.YELLOW) - color = cyan.blend(yellow, ratio) - - return color.hex - - -def topic_text(topic_name: str) -> Text: - """Format topic name with DimOS theme colors""" - if "#" in topic_name: - parts = topic_name.split("#", 1) - return Text(parts[0], style=theme.BRIGHT_WHITE) + Text("#" + parts[1], style=theme.BLUE) - - if topic_name[:4] == "/rpc": - return Text(topic_name[:4], style=theme.BLUE) + Text( - topic_name[4:], style=theme.BRIGHT_WHITE - ) - - return Text(topic_name, style=theme.BRIGHT_WHITE) - - -class LCMSpyApp(App): # type: ignore[type-arg] - """A real-time CLI dashboard for LCM traffic statistics using Textual.""" - - CSS_PATH = "../dimos.tcss" - - CSS = f""" - Screen {{ - layout: vertical; - background: {theme.BACKGROUND}; - }} - DataTable {{ - height: 2fr; - width: 1fr; - border: solid {theme.BORDER}; - background: {theme.BG}; - scrollbar-size: 0 0; - }} - DataTable > .datatable--header {{ - color: {theme.ACCENT}; - background: transparent; - }} - """ - - refresh_interval: float = 0.5 # seconds - - BINDINGS = [ - ("q", "quit"), - ("ctrl+c", "quit"), - ] - - def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def] - super().__init__(*args, **kwargs) - # Warn about missing system config before entering TUI raw mode. - from dimos.protocol.service.lcmservice import autoconf - - autoconf(check_only=True) - - self.spy = GraphLCMSpy(graph_log_window=0.5) - self.spy.start() - self.table: DataTable | None = None # type: ignore[type-arg] - - def compose(self) -> ComposeResult: - self.table = DataTable(zebra_stripes=False, cursor_type=None) # type: ignore[arg-type] - self.table.add_column("Topic") - self.table.add_column("Freq (Hz)") - self.table.add_column("Bandwidth") - self.table.add_column("Total Traffic") - yield self.table - - def on_mount(self) -> None: - self.set_interval(self.refresh_interval, self.refresh_table) - - async def on_unmount(self) -> None: - self.spy.stop() - - def refresh_table(self) -> None: - topics: list[SpyTopic] = list(self.spy.topic.values()) # type: ignore[arg-type, call-arg] - topics.sort(key=lambda t: t.total_traffic(), reverse=True) - self.table.clear(columns=False) # type: ignore[union-attr] - - for t in topics: - freq = t.freq(5.0) - kbps = t.kbps(5.0) - - self.table.add_row( # type: ignore[union-attr] - topic_text(t.name), - Text(f"{freq:.1f}", style=gradient(10, freq)), - Text(t.kbps_hr(5.0), style=gradient(1024 * 3, kbps)), - Text(t.total_traffic_hr()), - ) - - -def main() -> None: - import sys - - if len(sys.argv) > 1 and sys.argv[1] == "web": - import os - - from textual_serve.server import Server # type: ignore[import-not-found] - - server = Server(f"python {os.path.abspath(__file__)}") - server.serve() - else: - LCMSpyApp().run() - - -if __name__ == "__main__": - main() diff --git a/dimos/utils/cli/lcmspy/test_lcmspy.py b/dimos/utils/cli/lcmspy/test_lcmspy.py deleted file mode 100644 index 10799d8a9c..0000000000 --- a/dimos/utils/cli/lcmspy/test_lcmspy.py +++ /dev/null @@ -1,218 +0,0 @@ -# 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. - -import time - -import pytest - -from dimos.protocol.pubsub.impl.lcmpubsub import PickleLCM, Topic -from dimos.utils.cli.lcmspy.lcmspy import GraphLCMSpy, GraphTopic, LCMSpy, Topic as TopicSpy - - -@pytest.fixture -def pickle_lcm(): - lcm = PickleLCM() - lcm.start() - yield lcm - lcm.stop() - - -@pytest.fixture -def lcmspy_instance(): - spy = LCMSpy() - spy.start() - yield spy - spy.stop() - - -@pytest.fixture -def graph_lcmspy_instance(): - spy = GraphLCMSpy(graph_log_window=0.1) - spy.start() - time.sleep(0.2) # Wait for thread to start - yield spy - spy.stop() - - -def test_spy_basic(pickle_lcm, lcmspy_instance) -> None: - video_topic = Topic(topic="/video") - odom_topic = Topic(topic="/odom") - - for i in range(5): - pickle_lcm.publish(video_topic, f"video frame {i}") - time.sleep(0.1) - if i % 2 == 0: - pickle_lcm.publish(odom_topic, f"odometry data {i / 2}") - - # Wait a bit for messages to be processed - time.sleep(0.5) - - # Test statistics for video topic - video_topic_spy = lcmspy_instance.topic["/video"] - assert video_topic_spy is not None - - # Test frequency (should be around 10 Hz for 5 messages in ~0.5 seconds) - freq = video_topic_spy.freq(1.0) - assert freq > 0 - print(f"Video topic frequency: {freq:.2f} Hz") - - # Test bandwidth - kbps = video_topic_spy.kbps(1.0) - assert kbps > 0 - print(f"Video topic bandwidth: {kbps:.2f} kbps") - - # Test average message size - avg_size = video_topic_spy.size(1.0) - assert avg_size > 0 - print(f"Video topic average message size: {avg_size:.2f} bytes") - - # Test statistics for odom topic - odom_topic_spy = lcmspy_instance.topic["/odom"] - assert odom_topic_spy is not None - - freq = odom_topic_spy.freq(1.0) - assert freq > 0 - print(f"Odom topic frequency: {freq:.2f} Hz") - - kbps = odom_topic_spy.kbps(1.0) - assert kbps > 0 - print(f"Odom topic bandwidth: {kbps:.2f} kbps") - - avg_size = odom_topic_spy.size(1.0) - assert avg_size > 0 - print(f"Odom topic average message size: {avg_size:.2f} bytes") - - print(f"Video topic: {video_topic_spy}") - print(f"Odom topic: {odom_topic_spy}") - - -def test_topic_statistics_direct() -> None: - """Test Topic statistics directly without LCM""" - - topic = TopicSpy("/test") - - # Add some test messages - test_data = [b"small", b"medium sized message", b"very long message for testing purposes"] - - for _i, data in enumerate(test_data): - topic.msg(data) - time.sleep(0.1) # Simulate time passing - - # Test statistics over 1 second window - freq = topic.freq(1.0) - kbps = topic.kbps(1.0) - avg_size = topic.size(1.0) - - assert freq > 0 - assert kbps > 0 - assert avg_size > 0 - - print(f"Direct test - Frequency: {freq:.2f} Hz") - print(f"Direct test - Bandwidth: {kbps:.2f} kbps") - print(f"Direct test - Avg size: {avg_size:.2f} bytes") - - -def test_topic_cleanup() -> None: - """Test that old messages are properly cleaned up""" - - topic = TopicSpy("/test") - - # Add a message - topic.msg(b"test message") - initial_count = len(topic.message_history) - assert initial_count == 1 - - # Simulate time passing by manually adding old timestamps - old_time = time.time() - 70 # 70 seconds ago - topic.message_history.appendleft((old_time, 10)) - - # Trigger cleanup - topic._cleanup_old_messages(max_age=60.0) - - # Should only have the recent message - assert len(topic.message_history) == 1 - assert topic.message_history[0][0] > time.time() - 10 # Recent message - - -def test_graph_topic_basic() -> None: - """Test GraphTopic basic functionality""" - topic = GraphTopic("/test_graph") - - # Add some messages and update graphs - topic.msg(b"test message") - topic.update_graphs(1.0) - - # Should have history data - assert len(topic.freq_history) == 1 - assert len(topic.bandwidth_history) == 1 - assert topic.freq_history[0] > 0 - assert topic.bandwidth_history[0] > 0 - - -def test_graph_lcmspy_basic(graph_lcmspy_instance) -> None: - """Test GraphLCMSpy basic functionality""" - # Simulate a message - graph_lcmspy_instance.msg("/test", b"test data") - time.sleep(0.5) # Wait for graph update — macOS needs longer for thread scheduling - - # Should create GraphTopic with history - topic = graph_lcmspy_instance.topic["/test"] - assert isinstance(topic, GraphTopic) - assert len(topic.freq_history) > 0 - assert len(topic.bandwidth_history) > 0 - - -def test_lcmspy_global_totals(lcmspy_instance) -> None: - """Test that LCMSpy tracks global totals as a Topic itself""" - # Send messages to different topics - lcmspy_instance.msg("/video", b"video frame data") - lcmspy_instance.msg("/odom", b"odometry data") - lcmspy_instance.msg("/imu", b"imu data") - - # Verify each test topic received exactly one message (ignore LCM discovery packets) - for t in ("/video", "/odom", "/imu"): - assert len(lcmspy_instance.topic[t].message_history) == 1 - - # Check global statistics - global_freq = lcmspy_instance.freq(1.0) - global_kbps = lcmspy_instance.kbps(1.0) - global_size = lcmspy_instance.size(1.0) - - assert global_freq > 0 - assert global_kbps > 0 - assert global_size > 0 - - print(f"Global frequency: {global_freq:.2f} Hz") - print(f"Global bandwidth: {lcmspy_instance.kbps_hr(1.0)}") - print(f"Global avg message size: {global_size:.0f} bytes") - - -def test_graph_lcmspy_global_totals(graph_lcmspy_instance) -> None: - """Test that GraphLCMSpy tracks global totals with history""" - # Send messages - graph_lcmspy_instance.msg("/video", b"video frame data") - graph_lcmspy_instance.msg("/odom", b"odometry data") - time.sleep(0.2) # Wait for graph update - - # Update global graphs - graph_lcmspy_instance.update_graphs(1.0) - - # Should have global history - assert len(graph_lcmspy_instance.freq_history) == 1 - assert len(graph_lcmspy_instance.bandwidth_history) == 1 - assert graph_lcmspy_instance.freq_history[0] > 0 - assert graph_lcmspy_instance.bandwidth_history[0] > 0 - - print(f"Global frequency history: {graph_lcmspy_instance.freq_history[0]:.2f} Hz") - print(f"Global bandwidth history: {graph_lcmspy_instance.bandwidth_history[0]:.2f} kB/s") diff --git a/dimos/utils/cli/spy/conftest.py b/dimos/utils/cli/spy/conftest.py new file mode 100644 index 0000000000..ae182fbfd1 --- /dev/null +++ b/dimos/utils/cli/spy/conftest.py @@ -0,0 +1,39 @@ +# 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. + +"""Shared fixtures for the spy tests.""" + +import logging + +import pytest + +# core.py logs through this stdlib logger name (setup_logger() derives it +# from the module's file path). +_CORE_LOGGER = "dimos/utils/cli/spy/core.py" + + +@pytest.fixture +def spy_warnings(caplog): + """Capture spy core log lines via ``caplog``. + + The dimos logger is structlog over a stdlib logger with + ``propagate=False``, so caplog's root-level handler never sees it. + """ + lg = logging.getLogger(_CORE_LOGGER) + lg.addHandler(caplog.handler) + caplog.set_level(logging.WARNING, logger=_CORE_LOGGER) + try: + yield caplog + finally: + lg.removeHandler(caplog.handler) diff --git a/dimos/utils/cli/spy/core.py b/dimos/utils/cli/spy/core.py new file mode 100644 index 0000000000..843cfb5032 --- /dev/null +++ b/dimos/utils/cli/spy/core.py @@ -0,0 +1,385 @@ +# 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. + +"""Transport-agnostic pubsub spy: topic discovery, rates, sizes, liveness. + +Docs: docs/usage/transports/index.md ("Inspecting traffic"). + +HARD CONSTRAINT: the spy never decodes message payloads. Sources tap the +raw-bytes pubsub layer (LCMPubSubBase, ZenohPubSubBase — beneath the encoder +mixins), so the hot path per message is (topic string, payload length, +timestamp) and nothing else. Message *types* are still visible because they +are embedded in the topic string ("/cmd_vel#geometry_msgs.Twist"). +""" + +from __future__ import annotations + +from collections import deque +import contextlib +from dataclasses import dataclass +import threading +import time +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable + +from dimos.utils.logging_config import setup_logger + +if TYPE_CHECKING: + from collections.abc import Callable, Sequence + + # The entire hot-path event: (topic string incl. '#type' suffix, wire payload length). + TapCallback = Callable[[str, int], None] + +logger = setup_logger() + +# Seconds of per-message history kept for windowed stats. +DEFAULT_HISTORY_WINDOW = 60.0 + + +@dataclass(frozen=True, slots=True) +class SpyKey: + """Identity of one spied topic: which transport saw it + its raw topic string.""" + + transport: str # SpySource.name, e.g. "lcm", "zenoh" + topic: str # raw transport topic, e.g. "/cmd_vel#geometry_msgs.Twist" + + +def split_type_suffix(topic: str) -> tuple[str, str | None]: + """Split a spied topic string into (base_topic, msg_type_name or None). + + Both LCM and zenoh sources deliver topics in the uniform str(Topic) form + "base#pkg.Msg" (zenoh keys are converted back by _key_expr_to_topic). + Render-time helper — never called on the hot path. + + >>> split_type_suffix("/cmd_vel#geometry_msgs.Twist") + ('/cmd_vel', 'geometry_msgs.Twist') + >>> split_type_suffix("/plain") + ('/plain', None) + """ + base, sep, suffix = topic.partition("#") + return (base, suffix) if sep else (topic, None) + + +@dataclass(frozen=True, slots=True) +class WindowStats: + """One consistent reading of a TopicStats: windowed rates + lifetime totals. + + Built in a single pass under the stats lock (see TopicStats.window_stats), + so a UI thread never sees values torn by a concurrent record(). + """ + + freq: float # messages per second over the window + bytes_per_sec: float # payload bytes per second over the window + total_bytes: int + total_msgs: int + last_seen: float | None + + +class TopicStats: + """Sliding-window traffic statistics for one spied topic. + + Records only (timestamp, nbytes) pairs — no payloads are retained. + Timestamps are passed in explicitly (callers use time.time(); tests inject + values), so all stats are deterministic functions of recorded data. + + Thread-safety: record() may be called from transport threads while readers + query from a UI thread; readers must go through window_stats() (or the + freq/bytes_per_sec wrappers), which take the lock. + """ + + total_bytes: int + total_msgs: int + last_seen: float | None # timestamp of newest recorded message, None if none yet + + def __init__(self, history_window: float = DEFAULT_HISTORY_WINDOW) -> None: + """history_window: seconds of per-message history kept for windowed stats. + + total_bytes/total_msgs/last_seen survive eviction; only windowed stats + (freq/bytes_per_sec) forget evicted messages. + """ + self.history_window = history_window + self._history: deque[tuple[float, int]] = deque() # (timestamp, nbytes) + self._lock = threading.Lock() + self.total_bytes = 0 + self.total_msgs = 0 + self.last_seen = None + + def record(self, nbytes: int, timestamp: float) -> None: + """Hot path: O(1) amortized append + eviction of entries older than history_window.""" + with self._lock: + self._history.append((timestamp, nbytes)) + self.total_bytes += nbytes + self.total_msgs += 1 + self.last_seen = timestamp + cutoff = timestamp - self.history_window + while self._history and self._history[0][0] < cutoff: + self._history.popleft() + + def window_stats(self, window: float, now: float) -> WindowStats: + """Rates over [now - window, now] plus totals, in one locked pass.""" + cutoff = now - window + count = 0 + nbytes = 0 + with self._lock: + for ts, n in self._history: + if ts >= cutoff: + count += 1 + nbytes += n + return WindowStats( + freq=count / window if count else 0.0, + bytes_per_sec=nbytes / window if count else 0.0, + total_bytes=self.total_bytes, + total_msgs=self.total_msgs, + last_seen=self.last_seen, + ) + + def freq(self, window: float, now: float) -> float: + """Messages per second over [now - window, now]. 0.0 if none.""" + return self.window_stats(window, now).freq + + def bytes_per_sec(self, window: float, now: float) -> float: + """Payload bytes per second over [now - window, now]. 0.0 if none.""" + return self.window_stats(window, now).bytes_per_sec + + +@runtime_checkable +class SpySource(Protocol): + """One transport's raw firehose feeding the spy. + + Invariants for implementations: + - tap() rides the raw-bytes bus's subscribe_all; delivery scope and + conflation follow that transport's semantics (LCM delivers every + message; zenoh's subscribe_all is latest-per-topic over dimos/**). + - The tap callback receives (topic_str, nbytes) where topic_str is the + uniform str(Topic) form and nbytes is the wire payload length. + - Never decodes payloads, never retains them past the callback. + - start() is required before tap() delivers; stop() releases the bus. + """ + + name: str + + def start(self) -> None: ... + + def stop(self) -> None: ... + + def tap(self, callback: Callable[[str, int], None]) -> Callable[[], None]: ... + + def subscribe_decoded( + self, topic: str, callback: Callable[[Any], None] + ) -> Callable[[], None]: ... + + +class LCMSpySource: + """Spy source over LCM, via LCMPubSubBase.subscribe_all (raw regex '.*'). + + Delivers raw channel strings incl. the '#pkg.Msg' suffix. Payloads are the + LCM-encoded bytes; nbytes = len(payload). + """ + + name = "lcm" + + def __init__(self, **lcm_kwargs: Any) -> None: + """lcm_kwargs forwarded to LCMPubSubBase (e.g. lcm_url).""" + # Inline import: an unavailable LCM backend must not break the other + # spy sources (see default_sources). + from dimos.protocol.pubsub.impl.lcmpubsub import LCMPubSubBase + + self._bus = LCMPubSubBase(**lcm_kwargs) + + def start(self) -> None: + self._bus.start() + + def stop(self) -> None: + self._bus.stop() + + def tap(self, callback: Callable[[str, int], None]) -> Callable[[], None]: + return self._bus.subscribe_all(lambda msg, topic: callback(str(topic), len(msg))) + + def subscribe_decoded(self, topic: str, callback: Callable[[Any], None]) -> Callable[[], None]: + """Opt-in per-topic decoded tap — OFF the spy hot path. Not implemented in v1.""" + raise NotImplementedError + + +class ZenohSpySource: + """Spy source over zenoh, via ZenohPubSubBase.subscribe_all. + + zenoh's subscribe_all is latest-per-topic over dimos/** (best-effort), so + same-topic bursts between drains conflate away. nbytes = payload length; + topics arrive as str(Topic) with the type suffix reconstructed from the key. + """ + + name = "zenoh" + + def __init__(self, **zenoh_kwargs: Any) -> None: + """zenoh_kwargs forwarded to ZenohPubSubBase (e.g. mode/connect/listen, session_pool).""" + # Inline import: an unavailable zenoh backend must not break the other + # spy sources (see default_sources). + from dimos.protocol.pubsub.impl.zenohpubsub import ZenohPubSubBase + + self._bus = ZenohPubSubBase(**zenoh_kwargs) + + def start(self) -> None: + self._bus.start() + + def stop(self) -> None: + self._bus.stop() + + def tap(self, callback: Callable[[str, int], None]) -> Callable[[], None]: + return self._bus.subscribe_all(lambda msg, topic: callback(str(topic), len(msg))) + + def subscribe_decoded(self, topic: str, callback: Callable[[Any], None]) -> Callable[[], None]: + """Opt-in per-topic decoded tap — OFF the spy hot path. Not implemented in v1.""" + raise NotImplementedError + + +class TransportSpy: + """Aggregates SpySources into per-(transport, topic) stats plus global totals. + + Owns the sources' lifecycle: start() starts every source and taps it; + stop() untaps and stops them. Stats rows appear lazily as topics are first + seen (a topic with no traffic since start is invisible — the spy observes, + it does not enumerate). + + Thread-safety: tap callbacks arrive on transport threads; snapshot() may be + called from any thread and returns a consistent view for rendering. + """ + + totals: TopicStats # all messages across all sources + + def __init__( + self, + sources: Sequence[SpySource] | None = None, + history_window: float = DEFAULT_HISTORY_WINDOW, + ) -> None: + """sources=None means default_sources().""" + self._sources = list(sources) if sources is not None else default_sources() + self._history_window = history_window + self._stats: dict[SpyKey, TopicStats] = {} + self._lock = threading.Lock() + self._untaps: list[Callable[[], None]] = [] + self._live: list[SpySource] = [] # sources currently started + tapped + self.totals = TopicStats(history_window=history_window) + + def _tap_callback(self, transport: str) -> Callable[[str, int], None]: + def on_message(topic: str, nbytes: int) -> None: + now = time.time() + key = SpyKey(transport, topic) + with self._lock: + stats = self._stats.get(key) + if stats is None: + stats = TopicStats(history_window=self._history_window) + self._stats[key] = stats + stats.record(nbytes, now) + self.totals.record(nbytes, now) + + return on_message + + def _start_one(self, source: SpySource) -> None: + """Start + tap one source; if tap fails, stop it before propagating.""" + source.start() + try: + untap = source.tap(self._tap_callback(source.name)) + except BaseException: + with contextlib.suppress(Exception): + source.stop() + raise + self._live.append(source) + self._untaps.append(untap) + + def start(self, best_effort: bool = False) -> None: + """Start and tap every source. + + Strict (default): all-or-nothing — if any source fails to start or tap, + roll back the ones already started and re-raise. + best_effort: a source that fails start()/tap() is warned and skipped + (its own partial start is undone); the survivors keep running. Raises + only if no source starts at all. + """ + try: + for source in self._sources: + try: + self._start_one(source) + except Exception as exc: + if not best_effort: + raise + logger.warning( + "Skipping spy transport that failed to start", + transport=source.name, + error=str(exc), + ) + except BaseException: + self.stop() # roll back everything started so far, then propagate + raise + if best_effort and not self._live: + raise RuntimeError( + f"no spy transports could start (tried: {', '.join(s.name for s in self._sources)})" + ) + + def stop(self) -> None: + """Untap and stop everything; an error in one source never skips the rest.""" + for untap in self._untaps: + try: + untap() + except Exception as exc: + logger.warning("Error untapping spy source", error=str(exc)) + self._untaps.clear() + for source in self._live: + try: + source.stop() + except Exception as exc: + logger.warning("Error stopping spy source", transport=source.name, error=str(exc)) + self._live.clear() + + def snapshot(self) -> dict[SpyKey, TopicStats]: + """Current per-topic stats, safe to iterate while messages keep arriving.""" + with self._lock: + return dict(self._stats) + + +SOURCE_FACTORIES: dict[str, Callable[[], SpySource]] = { + LCMSpySource.name: LCMSpySource, + ZenohSpySource.name: ZenohSpySource, +} +"""Known transports by name; each factory constructs its source only when called. + +v1: lcm + zenoh. SHM/ROS/DDS/Redis are future sources. +""" + + +def validate_transport_names(names: Sequence[str]) -> None: + """Raise ValueError if any name is not a known SOURCE_FACTORIES transport.""" + unknown = [n for n in names if n not in SOURCE_FACTORIES] + if unknown: + raise ValueError( + f"unknown transport(s) {', '.join(unknown)} — valid choices: " + f"{', '.join(SOURCE_FACTORIES)}" + ) + + +def default_sources() -> list[SpySource]: + """The spy observes ALL transports simultaneously, regardless of DIMOS_TRANSPORT. + + A backend that fails to construct (missing import, native init error, …) is + skipped with a warning, so the default spy degrades to whatever transports + are available. Requesting a transport explicitly (constructing its + SOURCE_FACTORIES entry) keeps the hard error. + """ + sources: list[SpySource] = [] + for name, factory in SOURCE_FACTORIES.items(): + try: + sources.append(factory()) + except Exception as exc: # degrade on any backend init failure, not just imports + logger.warning("Skipping unavailable spy transport", transport=name, error=str(exc)) + if not sources: + raise RuntimeError(f"no spy transports available (tried: {', '.join(SOURCE_FACTORIES)})") + return sources diff --git a/dimos/utils/cli/spy/run_spy.py b/dimos/utils/cli/spy/run_spy.py new file mode 100644 index 0000000000..e38644d05d --- /dev/null +++ b/dimos/utils/cli/spy/run_spy.py @@ -0,0 +1,263 @@ +# 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. + +"""`dimos spy` TUI: live table of all topics across all pubsub transports. + +Textual app (DataTable, 0.5s refresh, theme colors, 'q' to quit). One row per +(transport, topic) from TransportSpy.snapshot(): + + Transport | Topic | Type | Freq (Hz) | Bandwidth | Total | Age + +- Topic/Type come from split_type_suffix(); rows sort by total traffic. +- Age is seconds since TopicStats.last_seen (liveness: dims/greys out stale rows). +- `--transport lcm --transport zenoh` (repeatable) filters sources; default all. +- LCM system config warning: call lcmservice.autoconf(check_only=True) before + entering raw TUI mode. +""" + +from __future__ import annotations + +import sys +import time +from typing import Any + +from rich.text import Text +from textual.app import App, ComposeResult +from textual.color import Color +from textual.widgets import DataTable + +from dimos.utils.cli import theme +from dimos.utils.cli.spy.core import ( + SOURCE_FACTORIES, + SpyKey, + TransportSpy, + WindowStats, + default_sources, + split_type_suffix, + validate_transport_names, +) +from dimos.utils.human import human_bytes + +# Rows older than this (seconds since last message) render dimmed as "stale". +STALE_AGE = 3.0 +# Window for freq / bandwidth readouts. +STAT_WINDOW = 5.0 +# Values at which the freq / bandwidth cell colors saturate. +FREQ_GRADIENT_MAX_HZ = 10.0 +BANDWIDTH_GRADIENT_MAX_BPS = 3 * 1024 + + +def gradient(max_value: float, value: float) -> str: + """Gradient from cyan (low) to yellow (high) using DimOS theme colors.""" + ratio = min(value / max_value, 1.0) if max_value else 0.0 + cyan = Color.parse(theme.CYAN) + yellow = Color.parse(theme.YELLOW) + return cyan.blend(yellow, ratio).hex + + +def topic_text(base: str) -> Text: + """Format a base topic name with DimOS theme colors.""" + if base[:4] == "/rpc": + return Text(base[:4], style=theme.BLUE) + Text(base[4:], style=theme.BRIGHT_WHITE) + return Text(base, style=theme.BRIGHT_WHITE) + + +def _parse_transports(argv: list[str]) -> list[str] | None: + """Parse repeated --transport flags; None means all default sources. + + `dimos spy` accepts only `--transport ` (repeatable). Any other token + — a stray positional or unknown flag — is rejected with a clear error rather + than silently ignored. + """ + transports: list[str] = [] + extra: list[str] = [] + i = 0 + while i < len(argv): + arg = argv[i] + if arg == "--transport": + i += 1 + if i < len(argv): + transports.append(argv[i]) + else: + raise SystemExit("Error: --transport requires a transport name") + elif arg.startswith("--transport="): + transports.append(arg.split("=", 1)[1]) + else: + extra.append(arg) + i += 1 + if extra: + raise SystemExit( + f"Error: unexpected argument(s) {', '.join(extra)} — " + "`dimos spy` accepts only --transport (repeatable)." + ) + try: + validate_transport_names(transports) + except ValueError as exc: + raise SystemExit(f"Error: {exc}") from None + return transports or None + + +class SpyApp(App[None]): + """A real-time dashboard for all-transport pubsub traffic using Textual.""" + + CSS_PATH = "../dimos.tcss" + + CSS = f""" + Screen {{ + layout: vertical; + background: {theme.BACKGROUND}; + }} + DataTable {{ + height: 2fr; + width: 1fr; + border: solid {theme.BORDER}; + background: {theme.BG}; + scrollbar-size: 0 0; + }} + DataTable > .datatable--header {{ + color: {theme.ACCENT}; + background: transparent; + }} + """ + + refresh_interval: float = 0.5 # seconds + + BINDINGS = [ + ("q", "quit"), + ("ctrl+c", "quit"), + ] + + def __init__(self, spy: TransportSpy, **kwargs: Any) -> None: + """spy: an already-started TransportSpy; the caller owns its lifecycle.""" + super().__init__(**kwargs) + self.spy = spy + self.table: DataTable[Text] | None = None + + def compose(self) -> ComposeResult: + self.table = DataTable(cursor_type="none") + self.table.add_column("Transport") + self.table.add_column("Topic") + self.table.add_column("Type") + self.table.add_column("Freq (Hz)") + self.table.add_column("Bandwidth") + self.table.add_column("Total") + self.table.add_column("Age") + yield self.table + + def on_mount(self) -> None: + self.set_interval(self.refresh_interval, self.refresh_table) + + def refresh_table(self) -> None: + table = self.table + if table is None: + return + now = time.time() + snap = self.spy.snapshot() + rows: list[tuple[SpyKey, WindowStats]] = sorted( + ((key, stats.window_stats(STAT_WINDOW, now)) for key, stats in snap.items()), + key=lambda kv: kv[1].total_bytes, + reverse=True, + ) + table.clear(columns=False) + + for key, stats in rows: + base, msg_type = split_type_suffix(key.topic) + freq = stats.freq + bps = stats.bytes_per_sec + age = now - stats.last_seen if stats.last_seen is not None else None + stale = age is not None and age > STALE_AGE + + if stale: + # Liveness: dim the whole row for topics gone quiet. + table.add_row( + Text(key.transport, style=theme.DIM), + Text(base, style=theme.DIM), + Text(msg_type or "", style=theme.DIM), + Text(f"{freq:.1f}", style=theme.DIM), + Text(f"{human_bytes(bps)}/s", style=theme.DIM), + Text(human_bytes(stats.total_bytes), style=theme.DIM), + Text(f"{age:.0f}s", style=theme.DIM), + ) + continue + + age_str = f"{age:.1f}s" if age is not None else "-" + table.add_row( + Text(key.transport, style=theme.BLUE), + topic_text(base), + Text(msg_type or "", style=theme.BLUE), + Text(f"{freq:.1f}", style=gradient(FREQ_GRADIENT_MAX_HZ, freq)), + Text(f"{human_bytes(bps)}/s", style=gradient(BANDWIDTH_GRADIENT_MAX_BPS, bps)), + Text(human_bytes(stats.total_bytes)), + Text(age_str, style=theme.ACCENT), + ) + + +def start_spy(transports: list[str] | None = None) -> TransportSpy: + """Build sources for the requested transports and start a TransportSpy. + + Runs before the TUI enters raw mode, so autoconf and degrade warnings + print to a normal terminal. transports=None means every available + transport, best-effort: a backend that fails to construct or start is + skipped with a warning so the spy still shows the others. An explicit + transport list is strict: unknown names and construction/start failures + are hard errors. + """ + if transports is None: + sources = default_sources() + else: + validate_transport_names(transports) + # Construct only the requested sources: a filtered-out transport is + # never imported or instantiated, and an unavailable one that was + # explicitly requested stays a hard error. + sources = [SOURCE_FACTORIES[name]() for name in transports] + # Warn about missing system config before entering TUI raw mode (LCM only). + if any(s.name == "lcm" for s in sources): + from dimos.protocol.service.lcmservice import autoconf + + autoconf(check_only=True) + spy = TransportSpy(sources=sources) + spy.start(best_effort=transports is None) + return spy + + +def main() -> None: + """Entry point for `dimos spy` (argv: --transport filters).""" + spy = start_spy(_parse_transports(sys.argv[1:])) + try: + SpyApp(spy).run() + finally: + spy.stop() + + +def lcm_only_argv(args: list[str]) -> list[str]: + """Build the `spy` argv for the LCM-only entry points (`lcmspy` / `dimos lcmspy`). + + Rejects an explicit --transport override (these entry points can't choose + transports); all other args pass through to `spy`. + """ + if any(a == "--transport" or a.startswith("--transport=") for a in args): + raise SystemExit( + "Error: lcmspy is LCM-only; use `dimos spy --transport ...` to choose transports." + ) + return ["spy", "--transport", "lcm", *args] + + +def lcm_main() -> None: + """`lcmspy` console-script shim: the spy over the LCM source only.""" + sys.argv = lcm_only_argv(sys.argv[1:]) + main() + + +if __name__ == "__main__": + main() diff --git a/dimos/utils/cli/spy/test_run_spy.py b/dimos/utils/cli/spy/test_run_spy.py new file mode 100644 index 0000000000..bda5cf6112 --- /dev/null +++ b/dimos/utils/cli/spy/test_run_spy.py @@ -0,0 +1,196 @@ +# 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. + +"""CLI arg handling for `dimos spy`: --transport parsing, validation, LCM-only alias.""" + +import sys +from typing import Any + +import pytest + +from dimos.utils.cli.spy import run_spy +from dimos.utils.cli.spy.core import SOURCE_FACTORIES, TransportSpy +from dimos.utils.cli.spy.run_spy import _parse_transports, lcm_only_argv, start_spy + + +def test_parse_transports_default_is_none(): + assert _parse_transports([]) is None + + +def test_parse_transports_collects_repeated_flags(): + assert _parse_transports(["--transport", "lcm"]) == ["lcm"] + assert _parse_transports(["--transport=zenoh"]) == ["zenoh"] + assert _parse_transports(["--transport", "lcm", "--transport", "zenoh"]) == ["lcm", "zenoh"] + + +def test_parse_transports_rejects_unknown_transport(): + with pytest.raises(SystemExit, match="zneoh"): + _parse_transports(["--transport", "zneoh"]) + + +def test_parse_transports_error_lists_valid_choices(): + with pytest.raises(SystemExit, match="lcm"): + _parse_transports(["--transport=nope"]) + + +def test_parse_transports_rejects_unexpected_positional(): + # A stray positional must fail loudly, not be silently ignored. + with pytest.raises(SystemExit, match="unexpected"): + _parse_transports(["foo"]) + with pytest.raises(SystemExit, match="unexpected"): + _parse_transports(["--transport", "lcm", "foo"]) + + +def test_parse_transports_missing_value_errors(): + with pytest.raises(SystemExit, match="requires a transport name"): + _parse_transports(["--transport"]) + + +class _StubSource: + name = "zenoh" + + def __init__(self) -> None: + self.started = False + + def start(self) -> None: + self.started = True + + def stop(self) -> None: + self.started = False + + def tap(self, callback: Any) -> Any: + return lambda: None + + def subscribe_decoded(self, topic: str, callback: Any) -> Any: + raise NotImplementedError + + +@pytest.fixture +def spy_starter(request): + """start_spy wrapper whose spies are stopped on teardown, even on failure. + + A factory (not a started spy) so each test can patch SOURCE_FACTORIES + before construction happens. + """ + + def _start(transports: list[str] | None = None) -> TransportSpy: + spy = start_spy(transports) + request.addfinalizer(spy.stop) + return spy + + return _start + + +def test_start_spy_default_skips_unavailable_backend(monkeypatch, spy_warnings, spy_starter): + def unavailable(): + raise ImportError("lcm backend missing") + + created: list[_StubSource] = [] + + def stub_factory() -> _StubSource: + source = _StubSource() + created.append(source) + return source + + monkeypatch.setitem(SOURCE_FACTORIES, "lcm", unavailable) + monkeypatch.setitem(SOURCE_FACTORIES, "zenoh", stub_factory) + spy_starter() # no --transport: degrades to the available backend + assert [s.started for s in created] == [True] + assert "lcm" in spy_warnings.text + + +def test_start_spy_default_skips_non_import_backend_failure(monkeypatch, spy_warnings, spy_starter): + # Any construction failure (e.g. a native init error), not just ImportError, + # must degrade rather than kill the default spy. + def broken(): + raise RuntimeError("zenoh native init failed") + + created: list[_StubSource] = [] + + def stub_factory() -> _StubSource: + source = _StubSource() + created.append(source) + return source + + monkeypatch.setitem(SOURCE_FACTORIES, "lcm", broken) + monkeypatch.setitem(SOURCE_FACTORIES, "zenoh", stub_factory) + spy_starter() # no --transport: degrades past the broken backend + assert [s.started for s in created] == [True] + assert "lcm" in spy_warnings.text and "native init failed" in spy_warnings.text + + +def test_start_spy_explicit_unavailable_transport_is_hard_error(monkeypatch): + def unavailable(): + raise RuntimeError("zenoh backend missing") + + monkeypatch.setitem(SOURCE_FACTORIES, "zenoh", unavailable) + with pytest.raises(RuntimeError, match="zenoh backend missing"): + start_spy(transports=["zenoh"]) + + +class _FailingStartSource(_StubSource): + name = "zenoh" + + def start(self) -> None: + raise RuntimeError("zenoh start failed") + + +def test_start_spy_default_survives_backend_start_failure(monkeypatch, spy_warnings, spy_starter): + # A backend that constructs but dies during start() must not kill the + # default spy — the survivor keeps running. + survivors: list[_StubSource] = [] + + def good_factory() -> _StubSource: + source = _StubSource() + source.name = "lcm" + survivors.append(source) + return source + + monkeypatch.setitem(SOURCE_FACTORIES, "lcm", good_factory) + monkeypatch.setitem(SOURCE_FACTORIES, "zenoh", _FailingStartSource) + spy_starter() # default path: best-effort + assert survivors and survivors[0].started + assert "zenoh" in spy_warnings.text + + +def test_start_spy_explicit_start_failure_is_hard_error(monkeypatch): + monkeypatch.setitem(SOURCE_FACTORIES, "zenoh", _FailingStartSource) + with pytest.raises(RuntimeError, match="zenoh start failed"): + start_spy(transports=["zenoh"]) # explicit: strict, no degrade + + +def test_lcm_only_argv_forwards_plain_args(): + assert lcm_only_argv([]) == ["spy", "--transport", "lcm"] + assert lcm_only_argv(["--foo", "bar"]) == ["spy", "--transport", "lcm", "--foo", "bar"] + + +def test_lcm_only_argv_rejects_transport_override(): + with pytest.raises(SystemExit, match="LCM-only"): + lcm_only_argv(["--transport", "zenoh"]) + with pytest.raises(SystemExit, match="LCM-only"): + lcm_only_argv(["--transport=zenoh"]) + + +def test_lcm_main_forwards_args_to_spy(monkeypatch): + seen: list[str] = [] + monkeypatch.setattr(run_spy, "main", lambda: seen.extend(sys.argv)) + monkeypatch.setattr(sys, "argv", ["lcmspy", "--foo"]) + run_spy.lcm_main() + assert seen == ["spy", "--transport", "lcm", "--foo"] + + +def test_lcm_main_rejects_transport_override(monkeypatch): + monkeypatch.setattr(sys, "argv", ["lcmspy", "--transport", "zenoh"]) + with pytest.raises(SystemExit, match="LCM-only"): + run_spy.lcm_main() diff --git a/dimos/utils/cli/spy/test_spy.py b/dimos/utils/cli/spy/test_spy.py new file mode 100644 index 0000000000..efe694398e --- /dev/null +++ b/dimos/utils/cli/spy/test_spy.py @@ -0,0 +1,451 @@ +# 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. + +"""Contract tests for the spy core (`dimos spy`). + +- TopicStats: deterministic windowed stats from injected timestamps. +- subscribe_all: LCM delivers every message (non-conflating regex '.*'). +- SpySources: count every message without ever decoding a payload. +- TransportSpy: merges sources into (transport, topic)-keyed stats + totals. +- subscribe_decoded: spec'd hook, must stay NotImplementedError in v1. +""" + +from collections.abc import Callable +import threading +import time +from typing import Any + +import pytest + +from dimos.msgs.geometry_msgs.Vector3 import Vector3 +from dimos.protocol.pubsub.impl.lcmpubsub import LCMPubSubBase, Topic +from dimos.protocol.pubsub.impl.zenohpubsub import ZenohPubSubBase +from dimos.protocol.service.zenohservice import ZenohSessionPool +from dimos.utils.cli.spy.core import ( + SOURCE_FACTORIES, + LCMSpySource, + SpyKey, + SpySource, + TopicStats, + TransportSpy, + ZenohSpySource, + default_sources, + split_type_suffix, +) + +VEC = Vector3(1.0, 2.0, 3.0) +VEC_BYTES = VEC.lcm_encode() + + +# TopicStats: pure, deterministic (injected timestamps, no sleeps) + + +def test_topic_stats_counts_and_totals(): + s = TopicStats(history_window=60.0) + t0 = 1000.0 + for i in range(10): + s.record(100, t0 + i * 0.1) + assert s.total_msgs == 10 + assert s.total_bytes == 1000 + assert s.last_seen == pytest.approx(t0 + 0.9) + + +def test_topic_stats_windowed_rates(): + s = TopicStats(history_window=60.0) + t0 = 1000.0 + for i in range(10): # 1 Hz, 50 bytes each + s.record(50, t0 + i) + now = t0 + 9.0 + assert s.freq(5.0, now) == pytest.approx(1.0, rel=0.25) + assert s.bytes_per_sec(5.0, now) == pytest.approx(50.0, rel=0.25) + + +def test_topic_stats_window_stats_single_reading(): + s = TopicStats(history_window=60.0) + t0 = 1000.0 + for i in range(10): # 1 Hz, 50 bytes each + s.record(50, t0 + i) + view = s.window_stats(5.0, now=t0 + 9.0) + assert view.freq == pytest.approx(1.0, rel=0.25) + assert view.bytes_per_sec == pytest.approx(50.0, rel=0.25) + assert view.total_bytes == 500 + assert view.total_msgs == 10 + assert view.last_seen == pytest.approx(t0 + 9.0) + + +def test_topic_stats_empty(): + s = TopicStats() + assert s.total_msgs == 0 + assert s.total_bytes == 0 + assert s.last_seen is None + assert s.freq(5.0, now=123.0) == 0.0 + assert s.bytes_per_sec(5.0, now=123.0) == 0.0 + + +def test_topic_stats_history_eviction_keeps_totals(): + s = TopicStats(history_window=60.0) + s.record(100, 0.0) + s.record(100, 1000.0) # first record is now far outside the history window + assert s.freq(60.0, now=1000.0) == pytest.approx(1 / 60.0) + assert s.total_msgs == 2 # totals survive eviction + assert s.total_bytes == 200 + + +# split_type_suffix: render-time topic parsing + + +def test_split_type_suffix(): + assert split_type_suffix("/cmd_vel#geometry_msgs.Twist") == ("/cmd_vel", "geometry_msgs.Twist") + assert split_type_suffix("dimos/cmd_vel#geometry_msgs.Twist") == ( + "dimos/cmd_vel", + "geometry_msgs.Twist", + ) + assert split_type_suffix("/plain") == ("/plain", None) + + +# subscribe_all contract: EVERY message, no conflation (spec-level fix) + + +@pytest.fixture +def lcm_bus(): + bus = LCMPubSubBase() + bus.start() + yield bus, Topic("/spy_contract", Vector3) + bus.stop() + + +def _publish_probes_until(pub, probe_topic: Topic, established: Callable[[], bool]) -> None: + """Publish probes until the subscription delivers one, instead of a fixed sleep.""" + deadline = time.time() + 10.0 + while not established(): + assert time.time() < deadline, f"subscription never saw a probe on {probe_topic}" + pub.publish(probe_topic, VEC_BYTES) + time.sleep(0.01) + + +def test_subscribe_all_delivers_every_message(lcm_bus): + """A gated (slow) consumer must still receive every message eventually. + + LCM's subscribe_all is non-conflating (regex '.*', 10k queue), so all 50 + messages are delivered even when the consumer lags the burst. + """ + n = 50 + bus, topic = lcm_bus + probe_topic = Topic("/spy_contract_probe", Vector3) + probe_seen = threading.Event() + gate = threading.Event() + got = [] + done = threading.Event() + + def cb(msg, t): + if str(t) == str(probe_topic): + probe_seen.set() + return + gate.wait(15.0) # simulate a consumer slower than the burst + got.append((str(t), len(msg))) + if len(got) >= n: + done.set() + + bus.subscribe_all(cb) + _publish_probes_until(bus, probe_topic, probe_seen.is_set) + + for _ in range(n): + bus.publish(topic, VEC_BYTES) + gate.set() + + done.wait(10.0) + ours = [g for g in got if g[0] == str(topic)] + assert len(ours) == n + assert all(size == len(VEC_BYTES) for _, size in ours) + + +# SpySources: every message counted, payloads never decoded + + +class _TapCollector: + def __init__(self, topic_str: str, n: int): + self.topic_str = topic_str + self.n = n + self.events: list[tuple[str, int]] = [] + self.done = threading.Event() + + def __call__(self, topic: str, nbytes: int) -> None: + self.events.append((topic, nbytes)) + if len(self.ours()) >= self.n: + self.done.set() + + def ours(self) -> list[tuple[str, int]]: + return [e for e in self.events if e[0] == self.topic_str] + + def saw(self, topic_str: str) -> bool: + return any(t == topic_str for t, _ in self.events) + + +def _assert_no_decode(monkeypatch): + """Make any Vector3 decode explode — the spy must never trigger one.""" + + def boom(*a, **k): + raise AssertionError("spy decoded a payload on the hot path") + + monkeypatch.setattr(Vector3, "lcm_decode", staticmethod(boom)) + + +@pytest.fixture +def lcm_spy_source(): + src = LCMSpySource() + src.start() + yield src + src.stop() + + +@pytest.fixture +def lcm_pub(): + pub = LCMPubSubBase() + pub.start() + yield pub + pub.stop() + + +def test_lcm_source_counts_all_without_decoding(monkeypatch, lcm_spy_source, lcm_pub): + _assert_no_decode(monkeypatch) + topic = Topic("/spy_e2e", Vector3) + probe = Topic("/spy_e2e_probe", Vector3) + collector = _TapCollector(str(topic), 20) + + lcm_spy_source.tap(collector) + _publish_probes_until(lcm_pub, probe, lambda: collector.saw(str(probe))) + for _ in range(20): + lcm_pub.publish(topic, VEC_BYTES) + time.sleep(0.01) + assert collector.done.wait(10.0) + assert [n for _, n in collector.ours()] == [len(VEC_BYTES)] * 20 + + +def test_lcm_source_counts_undecodable_garbage(monkeypatch, lcm_spy_source, lcm_pub): + """Payloads that would crash a decoder must still be counted (proves raw tap).""" + _assert_no_decode(monkeypatch) + topic = Topic("/spy_garbage", Vector3) + probe = Topic("/spy_garbage_probe", Vector3) + collector = _TapCollector(str(topic), 5) + + lcm_spy_source.tap(collector) + _publish_probes_until(lcm_pub, probe, lambda: collector.saw(str(probe))) + for _ in range(5): + lcm_pub.publish(topic, b"\x00garbage-not-lcm") + time.sleep(0.01) + assert collector.done.wait(10.0) + assert [n for _, n in collector.ours()] == [len(b"\x00garbage-not-lcm")] * 5 + + +@pytest.fixture +def zenoh_pool(): + pool = ZenohSessionPool() + yield pool + pool.close_all() + + +@pytest.fixture +def zenoh_spy_source(zenoh_pool): + src = ZenohSpySource(session_pool=zenoh_pool) + src.start() + yield src + src.stop() + + +@pytest.fixture +def zenoh_pub(zenoh_pool): + pub = ZenohPubSubBase(session_pool=zenoh_pool) + pub.start() + yield pub + pub.stop() + + +def test_zenoh_source_counts_all_without_decoding(monkeypatch, zenoh_spy_source, zenoh_pub): + _assert_no_decode(monkeypatch) + topic = Topic("dimos/spy_e2e", Vector3) + probe = Topic("dimos/spy_e2e_probe", Vector3) + collector = _TapCollector(str(topic), 20) + + zenoh_spy_source.tap(collector) + _publish_probes_until(zenoh_pub, probe, lambda: collector.saw(str(probe))) + for _ in range(20): + zenoh_pub.publish(topic, VEC_BYTES) + time.sleep(0.01) + assert collector.done.wait(10.0) + assert [n for _, n in collector.ours()] == [len(VEC_BYTES)] * 20 + + +# TransportSpy: merging + totals + lifecycle (fake sources, deterministic) + + +class FakeSource: + def __init__(self, name: str): + self.name = name + self.started = False + self.taps: list[Callable[[str, int], None]] = [] + + def start(self) -> None: + self.started = True + + def stop(self) -> None: + self.started = False + + def tap(self, callback): + self.taps.append(callback) + + def untap(): + if callback in self.taps: + self.taps.remove(callback) + + return untap + + def subscribe_decoded(self, topic: str, callback: Callable[[Any], None]): + raise NotImplementedError + + def emit(self, topic: str, nbytes: int) -> None: + for cb in list(self.taps): + cb(topic, nbytes) + + +def test_transport_spy_merges_sources_and_totals(): + a, b = FakeSource("lcm"), FakeSource("zenoh") + spy = TransportSpy(sources=[a, b]) + spy.start() + assert a.started and b.started + + a.emit("/t#geometry_msgs.Twist", 10) + a.emit("/t#geometry_msgs.Twist", 10) + b.emit("dimos/t#geometry_msgs.Twist", 20) + a.emit("/u", 5) + + snap = spy.snapshot() + assert snap[SpyKey("lcm", "/t#geometry_msgs.Twist")].total_msgs == 2 + assert snap[SpyKey("lcm", "/t#geometry_msgs.Twist")].total_bytes == 20 + assert snap[SpyKey("zenoh", "dimos/t#geometry_msgs.Twist")].total_msgs == 1 + assert snap[SpyKey("lcm", "/u")].total_bytes == 5 + assert spy.totals.total_msgs == 4 + assert spy.totals.total_bytes == 45 + + spy.stop() + assert not a.started and not b.started + assert not a.taps and not b.taps # untapped on stop + + +def test_transport_spy_snapshot_is_stable_copy(): + a = FakeSource("lcm") + spy = TransportSpy(sources=[a]) + spy.start() + a.emit("/t", 1) + snap = spy.snapshot() + a.emit("/new_topic_after_snapshot", 1) + assert SpyKey("lcm", "/new_topic_after_snapshot") not in snap # snapshot doesn't mutate + assert SpyKey("lcm", "/new_topic_after_snapshot") in spy.snapshot() + spy.stop() + + +def test_transport_spy_start_failure_stops_started_sources(): + class FailingStartSource(FakeSource): + def start(self) -> None: + raise RuntimeError("no zenoh here") + + a, b = FakeSource("lcm"), FailingStartSource("zenoh") + spy = TransportSpy(sources=[a, b]) + with pytest.raises(RuntimeError, match="no zenoh here"): + spy.start() + assert not a.started # rolled back, not left running + assert not a.taps + + +def test_transport_spy_tap_failure_stops_started_sources(): + class FailingTapSource(FakeSource): + def tap(self, callback): + raise RuntimeError("tap exploded") + + a, b = FakeSource("lcm"), FailingTapSource("zenoh") + spy = TransportSpy(sources=[a, b]) + with pytest.raises(RuntimeError, match="tap exploded"): + spy.start() + assert not a.started and not b.started + assert not a.taps + + +def test_transport_spy_stop_continues_past_failing_source(spy_warnings): + class FailingStopSource(FakeSource): + def stop(self) -> None: + raise RuntimeError("stop exploded") + + a, b = FailingStopSource("lcm"), FakeSource("zenoh") + spy = TransportSpy(sources=[a, b]) + spy.start() + spy.stop() # must not raise, and must still stop the healthy source + assert not b.started + assert not a.taps and not b.taps + assert "stop exploded" in spy_warnings.text + + +class _FailingStartSource(FakeSource): + def start(self) -> None: + raise RuntimeError("backend down") + + +def test_transport_spy_best_effort_skips_failed_source(request, spy_warnings): + good, bad = FakeSource("lcm"), _FailingStartSource("zenoh") + spy = TransportSpy(sources=[good, bad]) + spy.start(best_effort=True) # degrade to the survivor instead of dying + request.addfinalizer(spy.stop) + assert good.started and good.taps # survivor is live and tapped + assert not bad.started + good.emit("/t", 5) + assert spy.snapshot()[SpyKey("lcm", "/t")].total_bytes == 5 + assert "zenoh" in spy_warnings.text # skipped source is warned about + + +def test_transport_spy_best_effort_raises_when_all_fail(): + spy = TransportSpy(sources=[_FailingStartSource("lcm"), _FailingStartSource("zenoh")]) + with pytest.raises(RuntimeError, match="no spy transports could start"): + spy.start(best_effort=True) + + +def test_default_sources_skips_unavailable_backend(monkeypatch, spy_warnings): + def unavailable(): + raise ImportError("zenoh backend missing") + + monkeypatch.setitem(SOURCE_FACTORIES, "lcm", lambda: FakeSource("lcm")) + monkeypatch.setitem(SOURCE_FACTORIES, "zenoh", unavailable) + sources = default_sources() + assert [s.name for s in sources] == ["lcm"] # degrades instead of crashing + assert "zenoh" in spy_warnings.text + + +def test_default_sources_errors_when_no_backend_available(monkeypatch): + def unavailable(): + raise ImportError("backend missing") + + for name in SOURCE_FACTORIES: + monkeypatch.setitem(SOURCE_FACTORIES, name, unavailable) + with pytest.raises(RuntimeError, match="no spy transports available"): + default_sources() + + +def test_fake_source_satisfies_protocol(): + assert isinstance(FakeSource("x"), SpySource) + + +# Lazy decode hook: spec'd now, implemented in a follow-up (stays off hot path) + + +def test_subscribe_decoded_is_not_implemented_in_v1(): + src = LCMSpySource() + with pytest.raises(NotImplementedError): + src.subscribe_decoded("/x#geometry_msgs.Vector3", lambda m: None) diff --git a/docs/usage/cli.md b/docs/usage/cli.md index dc4de5d3c9..33329c7feb 100644 --- a/docs/usage/cli.md +++ b/docs/usage/cli.md @@ -201,6 +201,16 @@ Print resolved GlobalConfig values and their sources. dimos show-config ``` +### `dimos spy` + +Universal transport spy: a live table of every topic on every pubsub transport (LCM, Zenoh, or both), with per-topic message rate, bandwidth, size, and liveness. + +```bash +dimos spy # everything, all transports +dimos spy --transport zenoh # filter to one transport (repeatable flag) +dimos lcmspy # deprecated alias for: dimos spy --transport lcm +``` + --- ## Agent & MCP Commands @@ -297,7 +307,7 @@ humancli ### `lcmspy` -Monitor LCM messages in real time. +Deprecated alias for `dimos spy --transport lcm` (the LCM-only view of the spy). Prefer [`dimos spy`](#dimos-spy). ```bash lcmspy diff --git a/docs/usage/transports/index.md b/docs/usage/transports/index.md index 06d0b62636..889881b4c1 100644 --- a/docs/usage/transports/index.md +++ b/docs/usage/transports/index.md @@ -248,11 +248,19 @@ Received: (480, 640, 3) See [Modules](/docs/usage/modules.md) for more on module architecture. -## Inspecting LCM traffic (CLI) +## Inspecting traffic (CLI) -`lcmspy` shows topic frequency/bandwidth stats: +`dimos spy` is the universal transport spy: one live view of every topic moving on every +DimOS pubsub transport — names, message rates, bandwidth, sizes, and liveness — whether the +system runs on LCM, Zenoh, or both. -![lcmspy](../assets/lcmspy.png) +```bash +dimos spy # everything, all transports +dimos spy --transport zenoh # filter to one transport (repeatable flag) +dimos lcmspy # deprecated alias for: dimos spy --transport lcm +``` + +![dimos spy](../assets/lcmspy.png) `dimos topic echo /topic` listens on typed channels like `/topic#pkg.Msg` and decodes automatically: diff --git a/pyproject.toml b/pyproject.toml index ed64897e50..ba02c39c6a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -152,7 +152,7 @@ dependencies = [ [project.scripts] -lcmspy = "dimos.utils.cli.lcmspy.run_lcmspy:main" +lcmspy = "dimos.utils.cli.spy.run_spy:lcm_main" agentspy = "dimos.utils.cli.agentspy.agentspy:main" humancli = "dimos.utils.cli.human.humanclianim:main" dimos = "dimos.robot.cli.dimos:cli_main"