fix more flakiness

[paul-nechifor]

Problem

Closes DIM-XXX

Solution

How to Test

Contributor License Agreement

• I have read and approved the CLA.

[greptile-apps]

Greptile Summary

This PR reduces flakiness in tests and keeps the Unitree DDS module importable without optional SDK packages.

• Extends the high-volume pub/sub wait loop to wait for the expected message count.
• Uses a longer stall window before ending the pub/sub test early.
• Moves Unitree SDK imports into G1HighLevelDdsSdk.start().
• Stores loco API IDs on the SDK instance after the lazy imports succeed.

Confidence Score: 5/5

This looks safe to merge.

• No blocking issues found in the changed code.

Important Files Changed

Filename	Overview
dimos/protocol/pubsub/test_spec.py	Updates the high-volume message test to wait for all expected messages or a longer delivery stall.
dimos/robot/unitree/g1/effectors/high_level/dds_sdk.py	Defers Unitree SDK imports and initializes loco API IDs during start().

_{Reviews (2): Last reviewed commit: "fix more flakiness" | Re-trigger Greptile}

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
2291	2	2289	164

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

dimos.codebase_checks.test_blueprint_kwargs::test_blueprint_atom_kwargs_match_module_config[unitree-g1-nav-onboard]

Stack Traces | 3.19s run time

blueprint_name = 'unitree-g1-nav-onboard'

    @pytest.mark.parametrize("blueprint_name", _blueprint_params())
    def test_blueprint_atom_kwargs_match_module_config(blueprint_name: str) -> None:
        """Fail when blueprint kwargs cannot be consumed by their target module."""
>       blueprint = _get_blueprint_or_skip(blueprint_name)

blueprint_name = 'unitree-g1-nav-onboard'

dimos/codebase_checks/test_blueprint_kwargs.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/codebase_checks/test_blueprint_kwargs.py:36: in _get_blueprint_or_skip
    return get_blueprint_by_name(blueprint_name)
        blueprint_name = 'unitree-g1-nav-onboard'
        message    = "Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz aft... 'pull', '--include', 'data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz']' returned non-zero exit status 1."
dimos/robot/get_all_blueprints.py:47: in get_blueprint_by_name
    module = __import__(module_path, fromlist=[attr])
        attr       = 'unitree_g1_nav_onboard'
        module_path = 'dimos.robot.unitree.g1.blueprints.navigation.unitree_g1_nav_onboard'
        name       = 'unitree-g1-nav-onboard'
.../blueprints/navigation/unitree_g1_nav_onboard.py:47: in <module>
    "paths_dir": str(G1_LOCAL_PLANNER_PRECOMPUTED_PATHS),
        FastLio2   = <class 'dimos.hardware.sensors.lidar.fastlio2.module.FastLio2'>
        G1         = G1Config(name='unitree_g1', model_path=PosixPath('.../unitree/g1/g1.urdf'), heig...sition=Vector([          0           0         1.2]), orientation=Quaternion(0.000000, 0.000000, 0.000000, 1.000000))})
        G1_LOCAL_PLANNER_PRECOMPUTED_PATHS = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_p...nner_precomputed_paths.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xffa314ef70d0>
        MovementManager = <class 'dimos.navigation.movement_manager.movement_manager.MovementManager'>
        __builtins__ = <builtins>
        __cached__ = '.../navigation/__pycache__/unitree_g1_nav_onboard.cpython-312.pyc'
        __doc__    = None
        __file__   = '.../work/dimos/dimos/.../blueprints/navigation/unitree_g1_nav_onboard.py'
        __loader__ = <_frozen_importlib_external.SourceFileLoader object at 0xffa314dc5970>
        __name__   = 'dimos.robot.unitree.g1.blueprints.navigation.unitree_g1_nav_onboard'
        __package__ = 'dimos.robot.unitree.g1.blueprints.navigation'
        __spec__   = ModuleSpec(name='dimos.robot.unitree.g1.blueprints.navigation.unitree_g1_nav_onboard', loader=<_frozen_importlib_exter...dc5970>, origin='.../work/dimos/dimos/.../blueprints/navigation/unitree_g1_nav_onboard.py')
        _unitree_g1_onboard = Blueprint(blueprints=(BlueprintAtom(kwargs={'host_ip': '192.168.123.164', 'lidar_ip': '192.168.123.120'}, module=<clas...ers': 12, 'robot_model': 'unitree_g1'}), remapping_map=mappingproxy({}), requirement_checks=(), configurator_checks=())
        annotations = _Feature((3, 7, 0, 'beta', 1), None, 16777216)
        autoconnect = <function autoconnect at 0xffa3f3f99620>
        cmu_nav_rerun_config = <function cmu_nav_rerun_config at 0xffa31311a020>
        create_cmu_nav = <function create_cmu_nav at 0xffa31311a2a0>
        g1_odometry_tf_override = <function g1_odometry_tf_override at 0xffa314e6ccc0>
        g1_static_robot = <function g1_static_robot at 0xffa314e6c400>
        global_config = GlobalConfig(robot_ip=None, robot_ips=None, unitree_aes_128_key=None, xarm7_ip=None, xarm6_ip=None, can_port=None, dev...e, obstacle_avoidance=True, detection_model='moondream', listen_host='127.0.0.1', dimsim_scene='apt', dimsim_port=8090)
        vis_module = <function vis_module at 0xffa317172b60>
dimos/utils/data.py:369: in __str__
    return str(self._ensure_downloaded())
        self       = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_p...nner_precomputed_paths.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xffa314ef70d0>
dimos/utils/data.py:347: in _ensure_downloaded
    cache = get_data(filename)
        cache      = None
        filename   = 'unitree_g1_local_planner_precomputed_paths'
        self       = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_p...nner_precomputed_paths.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xffa314ef70d0>
dimos/utils/data.py:304: in get_data
    archive_path = _decompress_archive(_pull_lfs_archive(archive_name))
        archive_name = 'unitree_g1_local_planner_precomputed_paths'
        data_dir   = PosixPath('.../dimos/dimos/data')
        file_path  = PosixPath('.../dimos/dimos/data/unitree_g1_local_planner_precomputed_paths')
        name       = 'unitree_g1_local_planner_precomputed_paths'
        nested_path = None
        path_parts = ('unitree_g1_local_planner_precomputed_paths',)
dimos/utils/data.py:248: in _pull_lfs_archive
    _lfs_pull(file_path, repo_root)
        file_path  = PosixPath('.../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz')
        filename   = 'unitree_g1_local_planner_precomputed_paths'
        repo_root  = PosixPath('.../work/dimos/dimos')
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

file_path = PosixPath('.../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz')
repo_root = PosixPath('.../work/dimos/dimos')

    def _lfs_pull(file_path: Path, repo_root: Path, *, retries: int = 2) -> None:
        relative_path = file_path.relative_to(repo_root)
    
        env = os.environ.copy()
        env["GIT_LFS_FORCE_PROGRESS"] = "1"
    
        last_err: subprocess.CalledProcessError | None = None
        for attempt in range(1, retries + 2):  # retries + 1 total attempts
            try:
                subprocess.run(
                    ["git", "lfs", "pull", "--include", str(relative_path)],
                    cwd=repo_root,
                    check=True,
                    env=env,
                )
                return
            except subprocess.CalledProcessError as e:
                last_err = e
                if attempt <= retries:
                    time.sleep(attempt)  # 1s, 2s backoff
    
>       raise RuntimeError(
            f"Failed to pull LFS file {file_path} after {retries + 1} attempts: {last_err}"
        )
E       RuntimeError: Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz after 3 attempts: Command '['git', 'lfs', 'pull', '--include', 'data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz']' returned non-zero exit status 1.

attempt    = 3
env        = {'ACCEPT_EULA': 'Y', 'ACTIONS_ID_TOKEN_REQUEST_TOKEN': 'eyJhbGciOiJSUzI1NiIsImtpZCI6IjM4ODI2YjE3LTZhMzAtNWY5Yi1iMTY5LT...-version=2.0', 'ACTIONS_ORCHESTRATION_ID': '53d10b42-eed6-4c8f-a80e-a0d15bd599f5.tests.ubuntu-24_04-arm_3_14_fal', ...}
file_path  = PosixPath('.../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz')
last_err   = CalledProcessError(1, ['git', 'lfs', 'pull', '--include', 'data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz'])
relative_path = PosixPath('data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz')
repo_root  = PosixPath('.../work/dimos/dimos')
retries    = 2

dimos/utils/data.py:216: RuntimeError

dimos.robot.test_all_blueprints::test_blueprint_is_valid[unitree-g1-nav-onboard]

Stack Traces | 3.21s run time

blueprint_name = 'unitree-g1-nav-onboard'

    @pytest.mark.parametrize("blueprint_name", UBUNTU_BLUEPRINTS)
    def test_blueprint_is_valid(blueprint_name: str) -> None:
        """Validate blueprints that should import on the ubuntu-latest runner."""
>       _check_blueprint(blueprint_name)

blueprint_name = 'unitree-g1-nav-onboard'

dimos/robot/test_all_blueprints.py:106: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/robot/test_all_blueprints.py:82: in _check_blueprint
    blueprint = get_blueprint_by_name(blueprint_name)
        blueprint_name = 'unitree-g1-nav-onboard'
        message    = "Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz aft... 'pull', '--include', 'data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz']' returned non-zero exit status 1."
dimos/robot/get_all_blueprints.py:47: in get_blueprint_by_name
    module = __import__(module_path, fromlist=[attr])
        attr       = 'unitree_g1_nav_onboard'
        module_path = 'dimos.robot.unitree.g1.blueprints.navigation.unitree_g1_nav_onboard'
        name       = 'unitree-g1-nav-onboard'
.../blueprints/navigation/unitree_g1_nav_onboard.py:47: in <module>
    "paths_dir": str(G1_LOCAL_PLANNER_PRECOMPUTED_PATHS),
        FastLio2   = <class 'dimos.hardware.sensors.lidar.fastlio2.module.FastLio2'>
        G1         = G1Config(name='unitree_g1', model_path=PosixPath('.../unitree/g1/g1.urdf'), heig...sition=Vector([          0           0         1.2]), orientation=Quaternion(0.000000, 0.000000, 0.000000, 1.000000))})
        G1_LOCAL_PLANNER_PRECOMPUTED_PATHS = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_p...nner_precomputed_paths.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xff20ed179950>
        MovementManager = <class 'dimos.navigation.movement_manager.movement_manager.MovementManager'>
        __builtins__ = <builtins>
        __cached__ = '.../navigation/__pycache__/unitree_g1_nav_onboard.cpython-312.pyc'
        __doc__    = None
        __file__   = '.../work/dimos/dimos/.../blueprints/navigation/unitree_g1_nav_onboard.py'
        __loader__ = <_frozen_importlib_external.SourceFileLoader object at 0xff20ed181820>
        __name__   = 'dimos.robot.unitree.g1.blueprints.navigation.unitree_g1_nav_onboard'
        __package__ = 'dimos.robot.unitree.g1.blueprints.navigation'
        __spec__   = ModuleSpec(name='dimos.robot.unitree.g1.blueprints.navigation.unitree_g1_nav_onboard', loader=<_frozen_importlib_exter...181820>, origin='.../work/dimos/dimos/.../blueprints/navigation/unitree_g1_nav_onboard.py')
        _unitree_g1_onboard = Blueprint(blueprints=(BlueprintAtom(kwargs={'host_ip': '192.168.123.164', 'lidar_ip': '192.168.123.120'}, module=<clas...ers': 12, 'robot_model': 'unitree_g1'}), remapping_map=mappingproxy({}), requirement_checks=(), configurator_checks=())
        annotations = _Feature((3, 7, 0, 'beta', 1), None, 16777216)
        autoconnect = <function autoconnect at 0xff21cb69d620>
        cmu_nav_rerun_config = <function cmu_nav_rerun_config at 0xff20ea48c0e0>
        create_cmu_nav = <function create_cmu_nav at 0xff20ed172840>
        g1_odometry_tf_override = <function g1_odometry_tf_override at 0xff20ea7eade0>
        g1_static_robot = <function g1_static_robot at 0xff20ea7ea520>
        global_config = GlobalConfig(robot_ip=None, robot_ips=None, unitree_aes_128_key=None, xarm7_ip=None, xarm6_ip=None, can_port=None, dev...e, obstacle_avoidance=True, detection_model='moondream', listen_host='127.0.0.1', dimsim_scene='apt', dimsim_port=8090)
        vis_module = <function vis_module at 0xff20ee4bc220>
dimos/utils/data.py:369: in __str__
    return str(self._ensure_downloaded())
        self       = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_p...nner_precomputed_paths.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xff20ed179950>
dimos/utils/data.py:347: in _ensure_downloaded
    cache = get_data(filename)
        cache      = None
        filename   = 'unitree_g1_local_planner_precomputed_paths'
        self       = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_p...nner_precomputed_paths.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xff20ed179950>
dimos/utils/data.py:304: in get_data
    archive_path = _decompress_archive(_pull_lfs_archive(archive_name))
        archive_name = 'unitree_g1_local_planner_precomputed_paths'
        data_dir   = PosixPath('.../dimos/dimos/data')
        file_path  = PosixPath('.../dimos/dimos/data/unitree_g1_local_planner_precomputed_paths')
        name       = 'unitree_g1_local_planner_precomputed_paths'
        nested_path = None
        path_parts = ('unitree_g1_local_planner_precomputed_paths',)
dimos/utils/data.py:248: in _pull_lfs_archive
    _lfs_pull(file_path, repo_root)
        file_path  = PosixPath('.../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz')
        filename   = 'unitree_g1_local_planner_precomputed_paths'
        repo_root  = PosixPath('.../work/dimos/dimos')
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

file_path = PosixPath('.../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz')
repo_root = PosixPath('.../work/dimos/dimos')

    def _lfs_pull(file_path: Path, repo_root: Path, *, retries: int = 2) -> None:
        relative_path = file_path.relative_to(repo_root)
    
        env = os.environ.copy()
        env["GIT_LFS_FORCE_PROGRESS"] = "1"
    
        last_err: subprocess.CalledProcessError | None = None
        for attempt in range(1, retries + 2):  # retries + 1 total attempts
            try:
                subprocess.run(
                    ["git", "lfs", "pull", "--include", str(relative_path)],
                    cwd=repo_root,
                    check=True,
                    env=env,
                )
                return
            except subprocess.CalledProcessError as e:
                last_err = e
                if attempt <= retries:
                    time.sleep(attempt)  # 1s, 2s backoff
    
>       raise RuntimeError(
            f"Failed to pull LFS file {file_path} after {retries + 1} attempts: {last_err}"
        )
E       RuntimeError: Failed to pull LFS file .../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz after 3 attempts: Command '['git', 'lfs', 'pull', '--include', 'data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz']' returned non-zero exit status 1.

attempt    = 3
env        = {'ACCEPT_EULA': 'Y', 'ACTIONS_ID_TOKEN_REQUEST_TOKEN': 'eyJhbGciOiJSUzI1NiIsImtpZCI6IjM4ODI2YjE3LTZhMzAtNWY5Yi1iMTY5LT...-version=2.0', 'ACTIONS_ORCHESTRATION_ID': '53d10b42-eed6-4c8f-a80e-a0d15bd599f5.tests.ubuntu-24_04-arm_3_14_fal', ...}
file_path  = PosixPath('.../dimos/data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz')
last_err   = CalledProcessError(1, ['git', 'lfs', 'pull', '--include', 'data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz'])
relative_path = PosixPath('data/.lfs/unitree_g1_local_planner_precomputed_paths.tar.gz')
repo_root  = PosixPath('.../work/dimos/dimos')
retries    = 2

dimos/utils/data.py:216: RuntimeError

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