feat: dimos spy universal transport spy over LCM + Zenoh#2734
Conversation
Design ratified in ticket f6c74d39 (subscribe_all non-conflating decision confirmed by Ivan). Task package for a worker to implement: - TASK.md: goal, decisions, implementation map, acceptance - docs/usage/transports/spy.md: design doc (why, architecture, decisions) - spec.py: subscribe_all contract tightened (every message, no conflation); SubscribeLatestMixin.subscribe_latest() stub = the explicit conflation opt-in - protocol/pubsub/spy.py: TopicStats / SpySource / LCMSpySource / ZenohSpySource / TransportSpy stubs (typed, NotImplementedError bodies) - utils/cli/spy/run_spy.py + `dimos spy` CLI registration (TUI stub) - test_spy.py: 17 contract tests, 15 failing until implemented; the zenoh subscribe_all conflation bug fails sharply (2 of 50 messages delivered) mypy clean (796 files); existing pubsub suites green.
Implement the spy core (TopicStats, LCM/Zenoh SpySources, TransportSpy) that taps the raw-bytes pubsub layer — per-message hot path is (topic, len, ts), never decoding a payload — plus the Textual TUI (dimos spy) and CLI wiring. Spec fix: subscribe_all is now non-conflating on every transport; conflation moves to the opt-in SubscribeLatestMixin.subscribe_latest() (lifted from zenoh, transport-agnostic). Zenoh subscribe_all widens dimos/** -> ** and keeps its drain-stop bookkeeping via _register/_unregister_drain_stop hooks. Rerun bridge migrates to subscribe_latest. dimos lcmspy becomes a deprecated alias for dimos spy --transport lcm. Also strip 14 decorative section-banner comments from the delivered test_spy.py to satisfy codebase_checks/test_no_sections (approved by brain; no logic change).
leshy
requested review from
mustafab0,
paul-nechifor and
spomichter
as code owners
leshy
changed the title
Greptile SummaryThis PR adds a universal pubsub spy for LCM and Zenoh. The main changes are:
Confidence Score: 4/5The new spy paths need fixes before merging. Filtered CLI runs can initialize the wrong transports, and startup or subscription failures can leave background threads running.
dimos/utils/cli/spy/run_spy.py, dimos/protocol/pubsub/spy.py, dimos/protocol/pubsub/spec.py, dimos/robot/cli/dimos.py Important Files Changed
Flowchart%%{init: {'theme': 'neutral'}}%%
flowchart TD
CLI[dimos spy / dimos lcmspy] --> App[SpyApp]
App --> Sources[Selected SpySources]
Sources --> LCM[LCMSpySource]
Sources --> Zenoh[ZenohSpySource]
LCM --> Spy[TransportSpy]
Zenoh --> Spy
Spy --> Stats[TopicStats by transport and topic]
Stats --> TUI[Textual DataTable]
Bridge[Rerun bridge] --> Latest[subscribe_latest]
Latest --> All[subscribe_all]
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
CLI[dimos spy / dimos lcmspy] --> App[SpyApp]
App --> Sources[Selected SpySources]
Sources --> LCM[LCMSpySource]
Sources --> Zenoh[ZenohSpySource]
LCM --> Spy[TransportSpy]
Zenoh --> Spy
Spy --> Stats[TopicStats by transport and topic]
Stats --> TUI[Textual DataTable]
Bridge[Rerun bridge] --> Latest[subscribe_latest]
Latest --> All[subscribe_all]
Reviews (1): Last reviewed commit: "chore: drop internal task package file b..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| def __init__(self, transports: list[str] | None = None, **kwargs) -> None: # type: ignore[no-untyped-def] | ||
| super().__init__(**kwargs) | ||
| sources = default_sources() | ||
| if transports is not None: | ||
| sources = [s for s in sources if s.name in transports] |
There was a problem hiding this comment.
SpyApp constructs every default source before applying --transport, so dimos spy --transport lcm still imports and instantiates Zenoh and dimos spy --transport zenoh still constructs LCM. On an LCM-only host without Zenoh available, or a Zenoh-only run with a bad LCM URL, the filtered command can crash before the requested source is selected.
| sources = default_sources() | ||
| if transports is not None: | ||
| sources = [s for s in sources if s.name in transports] |
There was a problem hiding this comment.
_parse_transports() accepts any string and this filter silently drops all sources when none match. A typo such as dimos spy --transport zneoh launches a working TUI with no sources and no warning, so the user sees an empty table forever instead of a clear CLI error.
There was a problem hiding this comment.
|
|
||
| sys.argv = ["lcmspy", *ctx.args] | ||
| lcmspy_main() | ||
| sys.argv = ["spy", "--transport", "lcm", *ctx.args] |
There was a problem hiding this comment.
The deprecated alias prepends --transport lcm but still forwards user arguments afterward. dimos lcmspy --transport zenoh becomes spy --transport lcm --transport zenoh, so the LCM-only alias observes both transports instead of preserving the old LCM-only behavior.
| def start(self) -> None: | ||
| for source in self._sources: | ||
| source.start() | ||
| self._untaps.append(source.tap(self._tap_callback(source.name))) |
There was a problem hiding this comment.
TransportSpy.start() starts and taps sources in sequence without rollback. If LCM starts successfully and Zenoh startup or tap registration raises, the exception escapes from SpyApp.__init__ before unmount cleanup exists, leaving the already-started LCM service thread and socket running in the failed process.
| if not self._register_drain_stop(stop_drain, thread): | ||
| return lambda: None | ||
| inner_unsub = self.subscribe_all(collect) |
There was a problem hiding this comment.
subscribe_latest() starts the drain thread before the underlying subscribe_all() succeeds. If Zenoh raises while declaring the wildcard subscriber, the exception escapes and no unsubscribe is returned, leaving the daemon drain thread blocked on wake.wait() for the rest of the process.
Codecov Report❌ Patch coverage is @@ Coverage Diff @@
## main #2734 +/- ##
==========================================
+ Coverage 72.35% 72.47% +0.11%
==========================================
Files 902 904 +2
Lines 81820 82253 +433
Branches 7396 7419 +23
==========================================
+ Hits 59201 59612 +411
- Misses 20639 20653 +14
- Partials 1980 1988 +8
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 4 files with indirect coverage changes 🚀 New features to boost your workflow:
|
2 participants
`dimos spy` shows every topic flowing on every DimOS pubsub transport — LCM and Zenoh simultaneously — with per-topic rate, bandwidth, average size, total traffic and liveness in a Textual TUI (same look as `dimos lcmspy`, which stays as a deprecated alias routing to the LCM-only view).
Design doc: `docs/usage/transports/spy.md`.
Key properties:
Testing: 17 new contract tests (`dimos/protocol/pubsub/test_spy.py`); existing `test_spec` / `test_zenohpubsub` / `test_lcmpubsub` / rerun bridge suites green; mypy clean; TUI smoke-tested against live LCM traffic.