feat(messages): add built-in CDR types alongside legacy messages

.gitignore

@@ -106,3 +106,6 @@ optuna.db
 
 # policy blobs are loaded from outside the repository
 *.bin
+
+# Prepared from schema-sources.json; keep downloaded schemas out of Git.
+/packages/dimos-generated/interfaces/

MANIFEST.in

@@ -47,3 +47,8 @@ prune dimos/web/command-center-extension
 
 global-exclude test_*.py
 global-exclude conftest.py
+
+# Built-in message inputs are prepared automatically during source builds; the sdist is self-contained.
+include packages/dimos-generated/schema-sources.json
+recursive-include packages/dimos-generated/custom *.msg LICENSE
+recursive-include packages/dimos-generated/interfaces *.msg *LICENSE*

dimos/control/connection/test_connection_module.py

@@ -20,6 +20,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Callable, Iterator
 import math
 import pickle
@@ -480,19 +481,48 @@ def test_commands_for_other_robots_do_not_feed_the_deadman(rig: Callable[..., Ri
     assert r.module.halts == 1
 
 
-def test_any_accepted_command_feeds_the_one_deadman(rig: Callable[..., Rig]) -> None:
-    """Base commands keep the module alive after the body's stop."""
+def test_any_accepted_command_feeds_the_one_deadman(
+    rig: Callable[..., Rig], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Base commands reset the body's deadline on the same controlled clock."""
     r = rig(BodyAndBase)
-    body = MotorCommandArray(q=[0.1, 0.2], dq=[0, 0], kp=[1, 1], kd=[1, 1], tau=[0, 0])
-    forward = Twist(Vector3(0.1, 0.0, 0.0), Vector3(0.0, 0.0, 0.0))
-    r.send("motor_command", body)
-    deadline = time.monotonic() + 4 * DEADMAN_S
-    while time.monotonic() < deadline:
-        r.send("base_command", forward)
-        time.sleep(DEADMAN_S / 5)
-    assert r.module.halts == 0
-    assert wait_until(lambda: r.module.status().deadman_fired)
-    assert r.module.halts == 1
+
+    async def check_deadman() -> None:
+        loop = asyncio.get_running_loop()
+        now = loop.time()
+        written = asyncio.Event()
+        halted = asyncio.Event()
+        with monkeypatch.context() as patch:
+            patch.setattr(loop, "time", lambda: now)
+            write = r.module.write
+            halt = r.module.halt
+
+            def record_write(values: Any) -> None:
+                write(values)
+                written.set()
+
+            def record_halt() -> None:
+                halt()
+                halted.set()
+
+            patch.setattr(r.module, "write", record_write)
+            patch.setattr(r.module, "halt", record_halt)
+            body = MotorCommandArray(q=[0.1, 0.2], dq=[0, 0], kp=[1, 1], kd=[1, 1], tau=[0, 0])
+            forward = Twist(Vector3(0.1, 0.0, 0.0), Vector3(0.0, 0.0, 0.0))
+            r.send("motor_command", body)
+            await written.wait()
+            for _ in range(8):
+                now += DEADMAN_S / 2
+                written.clear()
+                r.send("base_command", forward)
+                await written.wait()
+            assert r.module.halts == 0
+            now += 2 * DEADMAN_S
+            await halted.wait()
+            assert r.module.status().deadman_fired
+            assert r.module.halts == 1
+
+    r.module.spawn(check_deadman()).result(timeout=10)
 
 
 def test_a_failing_write_halts_shows_the_error_and_carries_on(

dimos/message_codegen/build.py

@@ -0,0 +1,122 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""DimOS source-build inputs; imported only by packaging and preparation tools."""
+
+from hashlib import sha256
+import os
+from pathlib import Path, PurePosixPath
+import shutil
+from tempfile import TemporaryDirectory
+from typing import Annotated, Literal
+
+from dimos_cdrgen.source_inputs import prepare_sources
+from pydantic import AfterValidator, BaseModel, ConfigDict, Field
+
+ROOT = Path(__file__).resolve().parents[2]
+PROJECT = ROOT / "packages/dimos-generated"
+
+
+def _relative_path(value: str) -> str:
+    path = PurePosixPath(value)
+    if not path.parts or path.is_absolute() or ".." in path.parts:
+        raise ValueError("Unsafe source input path")
+    return value
+
+
+RelativePath = Annotated[str, AfterValidator(_relative_path)]
+Sha256 = Annotated[str, Field(pattern=r"^[0-9a-f]{64}$")]
+
+
+class SourceFile(BaseModel):
+    model_config = ConfigDict(strict=True, extra="forbid")
+
+    path: RelativePath
+    sha256: Sha256
+
+
+class SourceArchive(BaseModel):
+    model_config = ConfigDict(strict=True, extra="forbid")
+
+    repository: str | None = None
+    revision: str | None = None
+    url: str
+    sha256: Sha256
+    prefix: RelativePath
+    files: dict[RelativePath, SourceFile]
+
+
+class SourceLock(BaseModel):
+    model_config = ConfigDict(strict=True, extra="forbid")
+
+    format: Literal[1]
+    sources: list[SourceArchive]
+
+
+def _matches(directory: Path, files: dict[str, str]) -> bool:
+    return all(
+        (directory / name).is_file()
+        and sha256((directory / name).read_bytes()).hexdigest() == digest
+        for name, digest in files.items()
+    )
+
+
+def prepare_builtin_sources(
+    cache: Path | None = None, *, offline: bool = False, project: Path = PROJECT
+) -> Path:
+    cache = cache or Path(
+        os.environ.get("DIMOS_MESSAGE_SOURCE_CACHE", ROOT / "build/builtin-source-cache")
+    )
+    lock = project / "schema-sources.json"
+    data = SourceLock.model_validate_json(lock.read_text())
+    upstream = {
+        name: entry.sha256 for source in data.sources for name, entry in source.files.items()
+    }
+    custom_root = project / "custom/dimos_msgs"
+    custom = {
+        "dimos_msgs/msg/" + path.name: sha256(path.read_bytes()).hexdigest()
+        for path in (custom_root / "msg").glob("*.msg")
+    }
+    custom["licenses/dimos_msgs-LICENSE"] = sha256(
+        (custom_root / "LICENSE").read_bytes()
+    ).hexdigest()
+    expected = {**upstream, **custom}
+    target = project / "interfaces"
+    actual = {str(path.relative_to(target)) for path in target.rglob("*") if path.is_file()}
+    if actual == expected.keys() and _matches(target, expected):
+        return target
+    # Stage both downloaded and local definitions before replacing previous inputs.
+    with TemporaryDirectory(prefix=".interfaces-", dir=project) as directory:
+        stage = Path(directory) / "inputs"
+        if _matches(target, upstream):
+            stage.mkdir()
+            for name in upstream:
+                output = stage / name
+                output.parent.mkdir(parents=True, exist_ok=True)
+                shutil.copyfile(target / name, output)
+        else:
+            prepare_sources(lock, stage, cache / "archives", offline=offline)
+        shutil.copytree(custom_root / "msg", stage / "dimos_msgs/msg")
+        (stage / "licenses").mkdir(exist_ok=True)
+        shutil.copyfile(custom_root / "LICENSE", stage / "licenses/dimos_msgs-LICENSE")
+        backup = Path(directory) / "previous"
+        if target.exists():
+            target.rename(backup)
+        try:
+            stage.rename(target)
+        except OSError:
+            if backup.exists():
+                backup.rename(target)
+            raise
+    return target