feat: --record taps every published topic into a memory store by spomichter · Pull Request #3710 · dimensionalOS/dimos

spomichter · 2026-08-27T00:33:41Z

Summary

Right now to record to memory requires Go2Module, need to specific and route IN topics. Prefer to simple to have some --record mcap or --record --record-topics={color_image, ...} to specify full recording

Taps into coordinator loop and streams on a single worker to memory Sqlite store. Will add mcap support in next PR

After short recording:

You run: uv run dimos --record --simulation run unitree-go2

After run:

uv run dimos mem summary recordings/20260826-203855-unitree-go2/memory.db 
                                    memory.db                                    
┏━━━━━━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Stream         ┃  Items ┃    Hz ┃ Start (UTC)         ┃ Duration ┃       Size ┃
┡━━━━━━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━┩
│ lidar          │     43 │   2.0 │ 2026-08-27 03:39:06 │    21.2s │   5.14 MiB │
│ color_image    │    376 │  17.3 │ 2026-08-27 03:39:06 │    21.6s │   4.26 MiB │
│ global_map     │      9 │   0.4 │ 2026-08-27 03:39:07 │    20.2s │   3.08 MiB │
│ global_costmap │      9 │   0.4 │ 2026-08-27 03:39:07 │    20.2s │ 421.03 KiB │
│ tf             │    981 │  45.2 │ 2026-08-27 03:39:06 │    21.7s │ 292.19 KiB │
│ cmd_vel        │  1,855 │  99.7 │ 2026-08-27 03:39:07 │    18.6s │ 101.45 KiB │
│ tele_cmd_vel   │  1,855 │  98.6 │ 2026-08-27 03:39:07 │    18.8s │ 101.45 KiB │
│ goal           │  1,855 │  98.9 │ 2026-08-27 03:39:07 │    18.8s │  94.20 KiB │
│ way_point      │  1,855 │  98.9 │ 2026-08-27 03:39:07 │    18.8s │  94.20 KiB │
│ odom           │    981 │  45.2 │ 2026-08-27 03:39:06 │    21.7s │  82.39 KiB │
│ stop_movement  │  1,855 │  98.9 │ 2026-08-27 03:39:07 │    18.7s │  16.30 KiB │
│ camera_info    │     24 │     — │ 2026-08-27 03:38:56 │     0.0s │   8.67 KiB │
│ nav_cmd_vel    │      3 │  32.8 │ 2026-08-27 03:39:28 │     0.1s │   168.00 B │
│ goal_request   │      1 │     — │ 2026-08-27 03:39:28 │     0.0s │    86.00 B │
│ path           │      2 │ 289.1 │ 2026-08-27 03:39:28 │     0.0s │    68.00 B │
│ goal_reached   │      1 │     — │ 2026-08-27 03:39:28 │     0.0s │     9.00 B │
├────────────────┼────────┼───────┼─────────────────────┼──────────┼────────────┤
│ total          │ 11,705 │       │                     │          │  13.66 MiB │
└────────────────┴────────┴───────┴─────────────────────┴──────────┴────────────┘

--record usage

dimos --record run unitree-go2                                  # every stream -> recordings/<run-id>/memory.db
dimos --record --record-topics color_image run unitree-go2      # one stream
dimos --record --record-topics lidar,odom,tf run unitree-go2    # comma-separated, no spaces, no braces
dimos --record --record-topics 'global_*' run unitree-go2       # fnmatch globs on the stream name
dimos mem summary recordings/<run-id>/memory.db

Slop:
dimos --record run <blueprint> records every topic on the transport bus, rosbag-style, to one recordings/<run-id>/memory.db (same run id as logs/). No recorder module to wire per robot.

GlobalConfig.record ("" | sqlite) and record_topics (comma globs on the topic slug, default *). Bare --record means sqlite, normalized in argv like --simulation (normalize_argv, one table entry per optional-value flag). mcap is the next value once the writer lands (follow-up PR).
dimos/memory/tap.py: dimos run wraps coordinator.loop() in recording(), which subscribes to all topics on the active transport's pubsub (same backend selection as the Rerun bridge: LCM, or Zenoh when --transport zenoh) and appends each decoded message under its topic slug (/lidar -> lidar). Single writer, single file. Off unless --record; off under --replay. Topics without a dimos message type are skipped.
Poses are not resolved at record time; tf is a topic like any other and dimos map pose-fill derives poses on read.
Explicit Recorder modules (Go2Memory / unitree-go2-memory, unitree-go2-mid360-record, unitree-g1-record, ...) are unchanged: recording specific topics as a module stays available alongside --record.

Caveats: SHM-only streams are not on the bus and are not recorded; Zenoh's subscribe_all is latest-per-topic (best effort), LCM is complete.

- GlobalConfig.record ('' | sqlite) and record_topics (comma globs, default *). Bare --record means sqlite (argv normalize, same as --simulation). - Out.publish appends to recordings/<run-id>/memory-<pid>.db per process (dimos/memory/tap.py); off unless --record, off in replay. - Go2Memory / unitree-go2-memory removed: unitree-go2 --record replaces them.

codecov · 2026-08-27T00:43:42Z

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
4857	2	4855	87

View the top 1 failed test(s) by shortest run time

dimos.e2e_tests.test_manipulation_planning_groups::test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76dff02d6c90>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76dfd0905300>

    def test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan one generated plan over both arms and dispatch through one trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76dff02d6c90>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76dfd0905300>

dimos/e2e_tests/test_manipulation_planning_groups.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76dff02d6c90>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76dfd0905300>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x76dfd09058a0>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76dff02d6c90>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x76dfd09058a0>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4337656.169775288
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x76dfd09058a0>
timeout    = 120.0

.../utils/testing/waiting.py:35: TimeoutError

View the full list of 1 ❄️ flaky test(s)

dimos.e2e_tests.test_manipulation_planning_groups::test_single_arm_plans_and_executes_through_control_coordinator

Flake rate in main: 7.69% (Passed 12 times, Failed 1 times)

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76dff0279af0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76dfd0904360>

    def test_single_arm_plans_and_executes_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan with one arm and execute through its trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76dff0279af0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76dfd0904360>

dimos/e2e_tests/test_manipulation_planning_groups.py:159: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76dff0279af0>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76dfd0904360>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x76dfd0904fe0>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76dff0279af0>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x76dfd0904fe0>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4337535.991575881
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x76dfd0904fe0>
timeout    = 120.0

.../utils/testing/waiting.py:35: TimeoutError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

… process Replaces the per-process Out.publish hook: dimos run subscribes to every topic on the active transport (same backend selection as the Rerun bridge) and appends to recordings/<run-id>/memory.db.

…y --record and the Rerun bridge

greptile-apps · 2026-08-27T03:51:18Z

Greptile Summary

This change adds opt-in recording of matching transport streams to a per-run SQLite store, including command-line configuration and background persistence so transport callbacks can return promptly.

Recording failures can prevent a run from exiting cleanly. If encoding or a SQLite append raises, the writer stops; once its queue fills, shutdown waits indefinitely while trying to enqueue the stop marker.

Confidence Score: 4/5

Not safe to merge until recorder shutdown remains reliable after a write failure.

An exception while persisting a recorded message can leave the background writer stopped and block process cleanup indefinitely.

Files Needing Attention: dimos/memory/tap.py

T-Rex Logs

What T-Rex did

Reproduced the deterministic TransportRecorder append-failure and observed a close hang.
Validated the second P1 finding via its finding-comment-proof.
Validated the contract behavior of the run, noting append_entered was true, writer_alive was false, queue_full was true, and close_returned was false; the traceback points to dimos/memory/tap.py:73 and the child process exits cleanly, so the test runner does not hang.

_{Ran code and verified through T-Rex}

Comments Outside Diff (1)

General comment

TransportRecorder.close can block forever after an append failure
- Bug
  - A raising stream append terminates the daemon record-writer thread. If the bounded queue is subsequently full, close() blocks indefinitely in _queue.put(None) because no consumer remains to make space.
- Cause
  - _drain calls stream.append(msg, ts=ts) without exception handling at dimos/memory/tap.py:73; close performs an unbounded sentinel enqueue at line 77 before joining the dead writer.
- Fix
  - Handle append failures in _drain so the writer remains able to drain and receive shutdown, and make shutdown resilient to a failed writer/full queue (for example, use a bounded/conditional sentinel enqueue with explicit failure handling).
_{Ran code and verified through T-Rex}

_{Reviews (2): Last reviewed commit: "record: warn on first and every 1000th d..." | Re-trigger Greptile}

…, not a bus guess - ModuleCoordinator.transports exposes the wired (name, type) -> Transport table. - recording(coordinator.transports) subscribes each stream on whatever carries it (LCM, Zenoh, SHM, ...); streams are named by their blueprint name. - transport_factory / rerun bridge back to main (pubsub_backend dropped).

…queue (drops counted)

jeff-hykin

two things

delete Subscribable import Transport: [one click fix]
add a throttled warning for dropping messages: [one click fix]

Co-authored-by: Paul Nechifor <paul@nechifor.net>

…col (#3717)

…classes Transport; test_normalize_argv typed

spomichter · 2026-08-27T19:22:25Z

@greptile final review

…lose() can't deadlock

…lash and spaces in globs

…moved)

… before build

spomichter mentioned this pull request Aug 27, 2026

feat: --record flag, mcap recording by default #3706

Closed

6 tasks

spomichter added 7 commits August 26, 2026 19:17

cli: normalize_argv with an optional-value flag table

ce53986

keep Go2Memory / unitree-go2-memory as an explicit recorder module

1c722b4

tap: use Topic.topic, sanitize slug, record only dimos message types

94061c7

record: one memory.db per run, written by a bus subscriber in the run…

bd1641b

… process Replaces the per-process Out.publish hook: dimos run subscribes to every topic on the active transport (same backend selection as the Rerun bridge) and appends to recordings/<run-id>/memory.db.

transport_factory.pubsub_backend(): one place to pick the bus; used b…

0577d29

…y --record and the Rerun bridge

drop two comments

3d37b48

drop comment

94c3a83

spomichter commented Aug 27, 2026

View reviewed changes

Comment thread dimos/memory/tap.py Outdated

spomichter marked this pull request as ready for review August 27, 2026 03:42

spomichter requested review from Dreamsorcerer, aclauer, leshy, mustafab0 and paul-nechifor as code owners August 27, 2026 03:42

Merge branch 'main' into feat/record

3c7b7d6

spomichter commented Aug 27, 2026

View reviewed changes

Comment thread dimos/core/transport_factory.py Outdated

greptile-apps Bot reviewed Aug 27, 2026

View reviewed changes

Comment thread dimos/memory/tap.py Outdated

spomichter added 2 commits August 26, 2026 20:54

record: writer thread with bounded queue; transport callbacks only en…

242d544

…queue (drops counted)

paul-nechifor reviewed Aug 27, 2026

View reviewed changes

Comment thread dimos/cli/test_dimos.py Outdated

jeff-hykin reviewed Aug 27, 2026

View reviewed changes

Comment thread dimos/memory/tap.py

jeff-hykin reviewed Aug 27, 2026

View reviewed changes

Comment thread dimos/memory/tap.py Outdated

jeff-hykin requested changes Aug 27, 2026

View reviewed changes

github-actions Bot added the ready-to-merge Required CI checks have passed on this PR label Aug 27, 2026

Update dimos/cli/test_dimos.py

d4a6c16

Co-authored-by: Paul Nechifor <paul@nechifor.net>

github-actions Bot removed the ready-to-merge Required CI checks have passed on this PR label Aug 27, 2026

github-actions Bot added the ready-to-merge Required CI checks have passed on this PR label Aug 27, 2026

Merge branch 'main' into feat/record

083e072

github-actions Bot removed the ready-to-merge Required CI checks have passed on this PR label Aug 27, 2026

jeff-hykin and others added 2 commits August 27, 2026 12:14

refactor(record): use Transport instead of a local Subscribable proto…

5db2523

…col (#3717)

record: warn on first and every 1000th dropped message; test fake sub…

2fe5180

…classes Transport; test_normalize_argv typed

Merge branch 'main' into feat/record

a5f5b5c

greptile-apps Bot reviewed Aug 27, 2026

View reviewed changes

Comment thread dimos/memory/tap.py Outdated

spomichter added 5 commits August 27, 2026 12:45

record: writer survives a failed append (log and keep draining), so c…

bfbd860

…lose() can't deadlock

record: warn when --record-topics matches nothing; tolerate leading s…

1628a90

…lash and spaces in globs

record: no matching stream is fatal (coordinator stopped, empty db re…

b980bc0

…moved)

record: validate --record-topics against the blueprint's stream names…

c2122ec

… before build

test: CliRunner bypasses normalize_argv, pass --record sqlite explicitly

35d1372

spomichter merged commit 4003790 into main Aug 27, 2026
28 of 30 checks passed

spomichter deleted the feat/record branch August 27, 2026 21:58

spomichter mentioned this pull request Aug 27, 2026

docs: --record / --record-topics #3737

Merged

1 task

Navigation Menu

Uh oh!

feat: --record taps every published topic into a memory store - #3710

feat: --record taps every published topic into a memory store#3710
spomichter merged 21 commits into
mainfrom
feat/record

spomichter commented Aug 27, 2026 •

edited

Loading

Uh oh!

codecov Bot commented Aug 27, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

greptile-apps Bot commented Aug 27, 2026 •

edited

Loading

T-Rex Logs

Comments Outside Diff (1)

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jeff-hykin left a comment •

edited

Loading

Uh oh!

spomichter commented Aug 27, 2026

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

spomichter commented Aug 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Uh oh!

codecov Bot commented Aug 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

❌ 2 Tests Failed:

Uh oh!

Uh oh!

Uh oh!

greptile-apps Bot commented Aug 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Greptile Summary

Confidence Score: 4/5

T-Rex Logs

Comments Outside Diff (1)

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jeff-hykin left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

spomichter commented Aug 27, 2026

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

spomichter commented Aug 27, 2026 •

edited

Loading

codecov Bot commented Aug 27, 2026 •

edited

Loading

greptile-apps Bot commented Aug 27, 2026 •

edited

Loading

jeff-hykin left a comment •

edited

Loading