feat: dimos spy - universal transport spy over LCM + Zenoh

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)",
+    ),
 ]
 
 

dimos/robot/cli/dimos.py

@@ -619,13 +619,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 <name>`.",
+            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})

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
 
 import pytest
@@ -21,6 +22,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(
@@ -160,3 +162,55 @@ class TestModule(Module):
         "      * testmodule.spam: str (default: eggs)",
         "",
     ]
+
+
+@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