[TomCC7]

Linear project: Message Encoding Refactor

Native message libraries now supply Python/C++/Rust values and codecs. This update removes the owned compatibility/field-codec adapters and partial ROSIDL vendor tree, migrates runtime consumers, and keeps external-message authoring on the shared dimos build / pip backend. CDR changes encoding; LCM/Zenoh transport remains.

Published HEAD: 236e58f9a29ec4938e04f0750d7fad6e0aa77b50. Base: cc/test/cdr-record-replay. Normal forward commits/integration merges preserve published ancestry. Parent-relative diff: 945 files changed, 19272 insertions(+), 108502 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 in progress. Full wheel matrix passed. Standalone codegen passed; Jazzy reference and the main relay-image build failed before their checks because Docker Hub returned HTTP 429. These external failures remain red; no waiver, credential change, or all-green claim.

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.

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

Legacy dependency retirement above recording acceptance. Normal merge propagation preserves this layer’s parent-relative patch.

Layer 7/8 of native stack #4446. Published HEAD: 2d7173467686b8cca4a072f205d0e208e7729f58. Base: cc/test/cdr-record-replay. 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): 10 files changed, 25 insertions(+), 517 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.

Replaces closed, unmerged #4365. Original refs/history remain preserved.

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
6152	4	6148	227

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

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 0xff5af7932b40>
bench_results = [], lcm_url = 'udpm://239.255.76.67:11145?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 0xff5adefcccc0>
lcm_url    = 'udpm://239.255.76.67:11145?ttl=0'
n          = 15
proto      = 'zenoh'
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0xff5af7932b40>

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

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

Stack Traces | 0.008s run time

proto = 'lcm'
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0xff5adf72aba0>
bench_results = [], lcm_url = 'udpm://239.255.76.67:11145?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 0xff5af78afa60>
lcm_url    = 'udpm://239.255.76.67:11145?ttl=0'
n          = 15
proto      = 'lcm'
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0xff5adf72aba0>

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

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

Stack Traces | 0.008s run time

retry_until = <function retry_until at 0xff5bb556c680>
collector = namespace(received=[], event=<threading.Event at 0xff5adef7c080: unset>, callback=<function collector.<locals>.callback at 0xff5adefcc5e0>)
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0xff5adef7c170>
codec_transports = [<dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0xff5af7883b30>]

    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 0xff5af7883b30>]
collector  = namespace(received=[], event=<threading.Event at 0xff5adef7c080: unset>, callback=<function collector.<locals>.callback at 0xff5adefcc5e0>)
retry_until = <function retry_until at 0xff5bb556c680>
session_pool = <dimos.protocol.service.zenohservice.ZenohSessionPool object at 0xff5adef7c170>
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 0xff5af7883b30>

.../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 0xff5adefcc7c0>
        deadline   = <threading.Event at 0xff5af7933a10: unset>
        event      = <threading.Event at 0xff5adef7c080: unset>
        interval   = 0.01
        timeout    = 2.0
        timer      = <Timer(Thread-6735, stopped 280765199675776)>
.../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 0xff5af7883b30>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0xff5af7883b30>
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 0xff5af7883b30>
stream     = None

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

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

Stack Traces | 0.069s run time

retry_until = <function retry_until at 0xff5bb556c680>
collector = namespace(received=[], event=<threading.Event at 0xff5adf75a360: unset>, callback=<function collector.<locals>.callback at 0xff5af78342c0>)
codec_transports = [<dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0xff5adf75a720>]
lcm_url = 'udpm://239.255.76.67:11145?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 0xff5adf75a720>]
collector  = namespace(received=[], event=<threading.Event at 0xff5adf75a360: unset>, callback=<function collector.<locals>.callback at 0xff5af78342c0>)
lcm_url    = 'udpm://239.255.76.67:11145?ttl=0'
retry_until = <function retry_until at 0xff5bb556c680>
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 0xff5adf75a720>

.../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 0xff5af7835760>
        deadline   = <threading.Event at 0xff5adf7252b0: unset>
        event      = <threading.Event at 0xff5adf75a360: unset>
        interval   = 0.01
        timeout    = 2.0
        timer      = <Timer(Thread-6732, stopped 280765199675776)>
.../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 0xff5adf75a720>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.protocol.pubsub.benchmark.tool_replay_bench.CompressedCodec object at 0xff5adf75a720>
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 0xff5adf75a720>
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.}