[TomCC7]

Linear project: Message Encoding Refactor

Preserves the existing offline recording converter while integrating the native message migration from #4371. Conversion now uses registry codecs/schema lookup, explicit native constructors and numeric NumPy arrays. All seven converter feature files are retained; no-overwrite publication, metadata/timestamps, JPEG bytes and SQLite vector preservation remain covered.

Published HEAD: 2f68cf86499e807b20176f1bcc1119aff7632139. Base: cc/refactor/cdr-legacy-retirement. Normal forward commits/integration merges preserve published ancestry. Parent-relative diff: 7 files changed, 1724 insertions(+), 9 deletions(-).

The latest repair preserves owned snapshots explicitly, handles NumPy arrays without ambiguous truth/equality, excludes class metadata from configuration fields, restores reference-typestore test isolation, and fixes remaining registry-codec consumers. The macOS TurboJPEG CI prerequisite from #4247 is included. Benchmark paths are unchanged.

Local validation: 86 previously failing nonbenchmark cases passed; 15 mapping/navigation cases passed against unchanged golden files, including forced-color CLI checks; all three mock-coordinator subprocess tests passed. The converter/migration suite passed 34 tests after integration. All 12 modified production files passed scoped strict typing, and changed-file repository hooks passed. Three local cross-instance LCM delivery checks still time out on this host; no host-network changes or weakened assertions were used.

Earlier retained acceptance covers shared cross-package Header identity/assignment, dependency schema/version conflicts, full MCAP closure, and Python/C++/Rust codec exchange. Python uses unmodified rosbags 0.11.0 and NumPy; its registry creates classes/codecs in memory, with no compiler/download/file generation at import. C++ distribution is source-only; prebuilt native SDK distribution is not claimed. The 19 precise native-library strict expected failures remain unchanged.

Exact-head main CI is terminal with cancelled aggregate status. Python 3.12 finished with 6402 passed, 80 skipped and four paused benchmark failures only; the other three Python jobs were cancelled by fail-fast. Native C++/Rust builds, the exact documented three-language coordinator example, Rust, lint and documentation checks passed. The full wheel matrix passed on Linux x86_64, Linux ARM64 and macOS ARM64. Lint and documentation checks passed. The web job passed 150 Deno tests, then failed resolving the relay base image because auth.docker.io returned HTTP 504. Earlier Jazzy/installation runs encountered Docker Hub HTTP 429; no repeated pull loop or credential change was introduced. Standalone-codegen success at preceding HEAD 5ee0e429fb32d25a94ef4c1b942669578b28547f is retained evidence, not an exact-current-HEAD run. Benchmark investigation remains paused and aggregate CI is not green.

External-message walkthrough: dimos build, pip install ., and pip install -e . share the project configuration/backend. Use matching proposal artifacts, not assumed public releases.

M20 proprietary SDK/access, default-host multicast, Foxglove sign-in, and human visual acceptance remain distinct external limitations. No hardware operation, PR merge, or package-registry release occurred.

Latest top-only repair fixes the converter documentation branding and intentional pre-output validation expression. All 36 converter/policy tests passed; the combined reference-decoder and mock-coordinator suite passed 89 tests.

Preserved earlier description (implementation notes superseded by the update above)

Problem

Reviewers need the generator, bulk message definitions/packages and transport integration in their respective stack layers.

Solution

Explicit legacy recording conversion: MCAP for recording payloads and SQLite for writable analysis/vector preservation. Normal merge propagation preserves converter behavior and this layer’s parent-relative patch.

Layer 8/8 of native stack #4446. Published HEAD: 080c22f8cfaaa529e315d770c14c0210575e0a59. Base: cc/refactor/cdr-legacy-retirement. Normal forward commits/merges; existing ancestry and reviews are preserved, with no force push.

Parent-relative git diff (binary lockfile excluded from textual line counts): 7 files changed, 1599 insertions(+), 9 deletions(-).

Layout-repair validation: unchanged namespace-policy test plus generator tests: 41 passed; ownership/project checks: 19 passed; clean source/wheel/sdist/editable installation with Cargo/rustc/CMake/C++ disabled: 1 passed. A fresh backend wheel has api modules/stubs and no stale package initializers. C++/Rust/schema outputs are byte-identical to the pre-layout validated outputs; final built-in generation --check passed. Scoped strict typing and hooks passed. These focused checks are not a complete new-HEAD CI pass.

Production Rust declarations now come from pinned ros2msg 0.5.3 in a normal Cargo build.rs; Python installation only packages its sources and does not invoke Cargo. C++ declaration and CDR field traversal use pinned upstream ROSIDL/FastRTPS templates, with documented strict-bool patches. Python uses rosbags-generated declarations and CDR traversal with scoped malformed-input checks; the separate handwritten wire-layout walker was removed. Default, bounds, ownership and Sequence compatibility adapters remain. PR1 core: ten files including both helpers total 1,380 physical lines versus the original eight-file 1,463. Including maintained typing stubs gives 1,493 -> 1,444, a net 49-line reduction (83 is core-only). Full-stack message_codegen maintained non-test sources total 2,573 lines including stubs; CLI/SDK sources elsewhere are outside that count. Vendored sources and generated/copied artifacts are disclosed separately. Runtime native consumers use the distributed message crate; the redundant four-file Python/Cargo bootstrap (126 lines) was removed. No generator code was hidden in uncounted helpers.

Exact-new-HEAD remote CI is pending; no reruns or manual dispatches were requested for this batch. Historical runs do not verify these commits. Benchmark investigation remains paused, without a waiver or weakened gate.

Complete proposal walkthrough. M20 SDK/access, host multicast and human viewer acceptance remain external; no hardware operation, registry release or merge is claimed.

Upstream generator contracts, compatibility adapters and maintenance counts.

Namespace repair: five nonempty upstream initializers and two matching typing stubs are relocated to ordinary api modules; the empty parser initializer is omitted. Imports/EmPy templates reference modules directly. Licenses and original upstream source paths/hashes remain recorded, with the relocation and omitted empty file documented in manifests. test_no_init_files is unchanged; no blanket vendor exemption was added. The earlier mypy vendor-discovery boundary remains separate, and first-party strict settings remain enabled. The coauthor policy check remains unresolved; no attribution removal/history rewrite, force push, benchmark gate change, rerun or manual CI dispatch was requested.

[codecov]

❌ 4 Tests Failed:

Tests completed	Failed	Passed	Skipped
6410	4	6406	92

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

dimos.protocol.pubsub.benchmark.test_replay_bench::test_benchmark_image_vs_compressed[lcm]

Stack Traces | 0.005s run time

proto = 'lcm'
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0x7f93edb411e0>
bench_results = [], lcm_url = 'udpm://239.255.76.67:11259?ttl=0'

    @pytest.mark.parametrize("proto", ["lcm", "zenoh"])
    def test_benchmark_image_vs_compressed(proto, session_pool, bench_results, lcm_url) -> None:
        """Old path (raw Image on the wire) vs CompressedCodec(CompressedImage), same transport."""
        frame = make_image()  # 720p, 2.76 MB raw
        n = 15
    
        def inner(topic: str, typ: type):
            if proto == "lcm":
                return LCMTransport(topic, typ, url=f"{lcm_url}&recv_buf_size={2 * 1024 * 1024}")
            return ZenohTransport(topic, typ, session_pool=session_pool)
    
>       raw_wire = len(frame.encode())
E       AttributeError: 'sensor_msgs__msg__Image' object has no attribute 'encode'

bench_results = []
frame      = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...0, data=array([  0,  23,  47, ..., 255, 155,  54], shape=(2764800,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')
inner      = <function test_benchmark_image_vs_compressed.<locals>.inner at 0x7f93edb59870>
lcm_url    = 'udpm://239.255.76.67:11259?ttl=0'
n          = 15
proto      = 'lcm'
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0x7f93edb411e0>

.../pubsub/benchmark/test_replay_bench.py:198: AttributeError

dimos.protocol.pubsub.benchmark.test_replay_bench::test_benchmark_image_vs_compressed[zenoh]

Stack Traces | 0.006s run time

proto = 'zenoh'
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0x7f93edb25570>
bench_results = [], lcm_url = 'udpm://239.255.76.67:11259?ttl=0'

    @pytest.mark.parametrize("proto", ["lcm", "zenoh"])
    def test_benchmark_image_vs_compressed(proto, session_pool, bench_results, lcm_url) -> None:
        """Old path (raw Image on the wire) vs CompressedCodec(CompressedImage), same transport."""
        frame = make_image()  # 720p, 2.76 MB raw
        n = 15
    
        def inner(topic: str, typ: type):
            if proto == "lcm":
                return LCMTransport(topic, typ, url=f"{lcm_url}&recv_buf_size={2 * 1024 * 1024}")
            return ZenohTransport(topic, typ, session_pool=session_pool)
    
>       raw_wire = len(frame.encode())
E       AttributeError: 'sensor_msgs__msg__Image' object has no attribute 'encode'

bench_results = []
frame      = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...0, data=array([  0,  23,  47, ..., 255, 155,  54], shape=(2764800,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')
inner      = <function test_benchmark_image_vs_compressed.<locals>.inner at 0x7f93eda1cf70>
lcm_url    = 'udpm://239.255.76.67:11259?ttl=0'
n          = 15
proto      = 'zenoh'
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0x7f93edb25570>

.../pubsub/benchmark/test_replay_bench.py:198: AttributeError

dimos.protocol.pubsub.benchmark.test_replay_bench::test_roundtrip_over_zenoh

Stack Traces | 0.006s run time

retry_until = <function retry_until at 0x7f960f494ca0>
collector = namespace(received=[], event=<threading.Event object at 0x7f93f4af3be0>, callback=<function collector.<locals>.callback at 0x7f93edb58c10>)
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0x7f93f4af07c0>
codec_transports = [<dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93f4af3070>]

    def test_roundtrip_over_zenoh(retry_until, collector, session_pool, codec_transports) -> None:
        t = CompressedCodec(
            ZenohTransport("dimos/test/codec_zenoh", CompressedImage, session_pool=session_pool)
        )
        codec_transports.append(t)
        t.subscribe(collector.callback)
        src = make_image(320, 240)
>       retry_until(collector.event, lambda: t.broadcast(None, src))

codec_transports = [<dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93f4af3070>]
collector  = namespace(received=[], event=<threading.Event object at 0x7f93f4af3be0>, callback=<function collector.<locals>.callback at 0x7f93edb58c10>)
retry_until = <function retry_until at 0x7f960f494ca0>
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0x7f93f4af07c0>
src        = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...60, data=array([  0,  23,  47, ..., 255, 128,   1], shape=(230400,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')
t          = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93f4af3070>

.../pubsub/benchmark/test_replay_bench.py:94: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../utils/testing/waiting.py:50: in retry_until
    action()
        action     = <function test_roundtrip_over_zenoh.<locals>.<lambda> at 0x7f93eda1c550>
        deadline   = <threading.Event object at 0x7f93edba1330>
        event      = <threading.Event object at 0x7f93f4af3be0>
        interval   = 0.01
        timeout    = 2.0
        timer      = <Timer(Thread-4880, stopped 140272944015040)>
.../pubsub/benchmark/test_replay_bench.py:94: in <lambda>
    retry_until(collector.event, lambda: t.broadcast(None, src))
        src        = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...60, data=array([  0,  23,  47, ..., 255, 128,   1], shape=(230400,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')
        t          = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93f4af3070>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93f4af3070>
stream = None
msg = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...60, data=array([  0,  23,  47, ..., 255, 128,   1], shape=(230400,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')

    def broadcast(self, stream: Stream[Image] | None, msg: Image | CompressedImage) -> None:
        compressed = (
            msg
            if isinstance(msg, CompressedImage)
            else CompressedImage(
                header=msg.header,
                format="jpeg",
>               data=np.asarray(image_to_jpeg(msg, quality=self.quality), dtype=np.uint8),
            )
        )
E       ValueError: invalid literal for int() with base 10: b'\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t\x08\n\x0c\x14\r\x0c\x0b\x0b\x0c\x19\x12\x13\x0f\x14\x1d\x1a\x1f\x1e\

msg        = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...60, data=array([  0,  23,  47, ..., 255, 128,   1], shape=(230400,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')
self       = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93f4af3070>
stream     = None

.../pubsub/benchmark/tool_replay_bench.py:78: ValueError

dimos.protocol.pubsub.benchmark.test_replay_bench::test_roundtrip_over_lcm

Stack Traces | 0.071s run time

retry_until = <function retry_until at 0x7f960f494ca0>
collector = namespace(received=[], event=<threading.Event object at 0x7f93edb19a50>, callback=<function collector.<locals>.callback at 0x7f93edb59900>)
codec_transports = [<dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93edba15a0>]
lcm_url = 'udpm://239.255.76.67:11259?ttl=0'

    def test_roundtrip_over_lcm(retry_until, collector, codec_transports, lcm_url) -> None:
        t = CompressedCodec(LCMTransport("dimos/test/codec_lcm", CompressedImage, url=lcm_url))
        codec_transports.append(t)
        t.subscribe(collector.callback)
        src = make_image(320, 240)
>       retry_until(collector.event, lambda: t.broadcast(None, src))

codec_transports = [<dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93edba15a0>]
collector  = namespace(received=[], event=<threading.Event object at 0x7f93edb19a50>, callback=<function collector.<locals>.callback at 0x7f93edb59900>)
lcm_url    = 'udpm://239.255.76.67:11259?ttl=0'
retry_until = <function retry_until at 0x7f960f494ca0>
src        = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...60, data=array([  0,  23,  47, ..., 255, 128,   1], shape=(230400,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')
t          = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93edba15a0>

.../pubsub/benchmark/test_replay_bench.py:79: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../utils/testing/waiting.py:50: in retry_until
    action()
        action     = <function test_roundtrip_over_lcm.<locals>.<lambda> at 0x7f93edb59240>
        deadline   = <threading.Event object at 0x7f93edb19a80>
        event      = <threading.Event object at 0x7f93edb19a50>
        interval   = 0.01
        timeout    = 2.0
        timer      = <Timer(Thread-4877, stopped 140272960796352)>
.../pubsub/benchmark/test_replay_bench.py:79: in <lambda>
    retry_until(collector.event, lambda: t.broadcast(None, src))
        src        = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...60, data=array([  0,  23,  47, ..., 255, 128,   1], shape=(230400,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')
        t          = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93edba15a0>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93edba15a0>
stream = None
msg = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...60, data=array([  0,  23,  47, ..., 255, 128,   1], shape=(230400,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')

    def broadcast(self, stream: Stream[Image] | None, msg: Image | CompressedImage) -> None:
        compressed = (
            msg
            if isinstance(msg, CompressedImage)
            else CompressedImage(
                header=msg.header,
                format="jpeg",
>               data=np.asarray(image_to_jpeg(msg, quality=self.quality), dtype=np.uint8),
            )
        )
E       ValueError: invalid literal for int() with base 10: b'\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t\x08\n\x0c\x14\r\x0c\x0b\x0b\x0c\x19\x12\x13\x0f\x14\x1d\x1a\x1f\x1e\

msg        = sensor_msgs__msg__Image(header=std_msgs__msg__Header(stamp=builtin_interfaces__msg__Time(sec=42, nanosec=125000000, __...60, data=array([  0,  23,  47, ..., 255, 128,   1], shape=(230400,), dtype=uint8), __msgtype__='sensor_msgs/msg/Image')
self       = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0x7f93edba15a0>
stream     = None

.../pubsub/benchmark/tool_replay_bench.py:78: ValueError

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