From 99f29f74be0862c25ae2f8fbcc6001b7b2903e10 Mon Sep 17 00:00:00 2001 From: Sam Bull Date: Thu, 30 Jul 2026 23:36:03 +0100 Subject: [PATCH 1/3] Temp fix for Mac CI --- .github/workflows/ci.yml | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 39af493963..21f3168e08 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -570,7 +570,7 @@ jobs: - os: macOS container: null # run on host — `container:` is Linux-only markers: "self_hosted" - experimental: true + experimental: false runs-on: - self-hosted - ${{ matrix.os }} @@ -668,6 +668,13 @@ jobs: - name: Fetch cmu_nav native modules from Cachix if: contains(matrix.markers, 'skipif_no_ros') run: uv run bin/build-cmu-nav-natives + - name: Enable zenoh network scouting + # Stopgap: loopback-pinned multicast scouting (the zenoh_scouting=off + # default since f85870726) never delivers on Darwin lo0, so the + # coordinator and workers can't discover each other and every RPC + # times out. Scout on the default interface until the pin is fixed. + if: matrix.os == 'macOS' + run: echo "DIMOS_ZENOH_SCOUTING=1" >> "$GITHUB_ENV" - name: Run tests run: uv run pytest --cov=dimos/ --junitxml=junit.xml -m '(${{ matrix.markers }}) and not mujoco' - name: Re-run the failing tests with maximum verbosity From 580fc4b0d0506a8b788fb99689efd43d267d3e8e Mon Sep 17 00:00:00 2001 From: Sam Bull Date: Thu, 30 Jul 2026 23:52:34 +0100 Subject: [PATCH 2/3] Fix --- .github/workflows/ci.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 21f3168e08..d4a06fb1f5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -673,8 +673,10 @@ jobs: # default since f85870726) never delivers on Darwin lo0, so the # coordinator and workers can't discover each other and every RPC # times out. Scout on the default interface until the pin is fixed. + # NB: GlobalConfig has no env_prefix; fields bind to bare env names + # (only `transport` aliases DIMOS_TRANSPORT). if: matrix.os == 'macOS' - run: echo "DIMOS_ZENOH_SCOUTING=1" >> "$GITHUB_ENV" + run: echo "ZENOH_SCOUTING=1" >> "$GITHUB_ENV" - name: Run tests run: uv run pytest --cov=dimos/ --junitxml=junit.xml -m '(${{ matrix.markers }}) and not mujoco' - name: Re-run the failing tests with maximum verbosity From 3d49999ec9b25a508b4c2ff5ac5321b8e5e1d111 Mon Sep 17 00:00:00 2001 From: Sam Bull Date: Fri, 31 Jul 2026 15:58:27 +0100 Subject: [PATCH 3/3] Proposed zenoh fix --- .github/workflows/ci.yml | 9 -- dimos/core/coordination/python_worker.py | 30 +++++- .../coordination/worker_manager_python.py | 41 ++++++-- dimos/protocol/service/zenohservice.py | 98 +++++++++++++++---- 4 files changed, 140 insertions(+), 38 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d4a06fb1f5..505cf13fed 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -668,15 +668,6 @@ jobs: - name: Fetch cmu_nav native modules from Cachix if: contains(matrix.markers, 'skipif_no_ros') run: uv run bin/build-cmu-nav-natives - - name: Enable zenoh network scouting - # Stopgap: loopback-pinned multicast scouting (the zenoh_scouting=off - # default since f85870726) never delivers on Darwin lo0, so the - # coordinator and workers can't discover each other and every RPC - # times out. Scout on the default interface until the pin is fixed. - # NB: GlobalConfig has no env_prefix; fields bind to bare env names - # (only `transport` aliases DIMOS_TRANSPORT). - if: matrix.os == 'macOS' - run: echo "ZENOH_SCOUTING=1" >> "$GITHUB_ENV" - name: Run tests run: uv run pytest --cov=dimos/ --junitxml=junit.xml -m '(${{ matrix.markers }}) and not mujoco' - name: Re-run the failing tests with maximum verbosity diff --git a/dimos/core/coordination/python_worker.py b/dimos/core/coordination/python_worker.py index 95bbd47703..e67a581618 100644 --- a/dimos/core/coordination/python_worker.py +++ b/dimos/core/coordination/python_worker.py @@ -37,6 +37,11 @@ ) from dimos.core.global_config import GlobalConfig, global_config from dimos.protocol.pubsub.impl.webrtc.providers.spec import shutdown_all_providers +from dimos.protocol.service.zenohservice import ( + ZenohConfig, + configure_zenoh_mesh, + default_session_pool, +) from dimos.utils.logging_config import setup_logger from dimos.utils.sequential_ids import SequentialIds @@ -169,6 +174,9 @@ def __init__(self) -> None: self._conn: Connection | None = None self._worker_id: int = _worker_ids.next() self.dedicated: bool = False + # Loopback locator the worker's zenoh sessions listen on (zenoh + # transport only); the coordinator and later-spawned workers dial it. + self.listen_endpoint: str | None = None @property def module_count(self) -> int: @@ -201,14 +209,16 @@ def reserve_slot(self) -> None: """Reserve a slot so _select_worker() sees the pending load.""" self._reserved += 1 - def start_process(self) -> None: + def start_process(self, zenoh_mesh: tuple[str, tuple[str, ...]] | None = None) -> None: ctx = get_forkserver_context() parent_conn, child_conn = ctx.Pipe() self._conn = parent_conn + if zenoh_mesh is not None: + self.listen_endpoint = zenoh_mesh[0] self._process = ctx.Process( target=_worker_entrypoint, - args=(child_conn, self._worker_id), + args=(child_conn, self._worker_id, zenoh_mesh), daemon=True, ) self._process.start() @@ -327,8 +337,22 @@ class _WorkerState: should_stop: bool = False -def _worker_entrypoint(conn: Connection, worker_id: int) -> None: +def _worker_entrypoint( + conn: Connection, + worker_id: int, + zenoh_mesh: tuple[str, tuple[str, ...]] | None = None, +) -> None: signal.signal(signal.SIGINT, signal.SIG_IGN) # coordinator handles shutdown + # Before any module can open a session: the worker's zenoh endpoints are + # fixed at spawn (its own listen port, the endpoints of older siblings). + # The session opens eagerly so the listener is up for siblings and the + # coordinator well before the first deploy; modules reuse it via the pool. + # Mesh args imply the host runs zenoh — sync before the config is built + # so its connect endpoints match the sessions modules acquire later. + if zenoh_mesh is not None: + configure_zenoh_mesh(*zenoh_mesh) + global_config.update(transport="zenoh") + default_session_pool.acquire(ZenohConfig()) state = _WorkerState(instances={}, worker_id=worker_id) try: diff --git a/dimos/core/coordination/worker_manager_python.py b/dimos/core/coordination/worker_manager_python.py index aa440a20b2..bf25bd9044 100644 --- a/dimos/core/coordination/worker_manager_python.py +++ b/dimos/core/coordination/worker_manager_python.py @@ -22,6 +22,7 @@ from dimos.core.global_config import GlobalConfig from dimos.core.module import ModuleBase, ModuleSpec from dimos.core.rpc_client import ModuleProxyProtocol, RPCClient +from dimos.protocol.service.zenohservice import allocate_mesh_endpoint, configure_zenoh_mesh from dimos.utils.logging_config import setup_logger from dimos.utils.safe_thread_map import safe_thread_map @@ -53,14 +54,35 @@ def __init__(self, g: GlobalConfig) -> None: self._started = False self._stats_monitor: StatsMonitor | None = None + def _mesh_endpoints(self) -> tuple[str, ...]: + return tuple(w.listen_endpoint for w in self._workers if w.listen_endpoint) + + def _spawn_worker(self) -> PythonWorker: + """Spawn a worker process wired into the explicit loopback zenoh mesh. + + Sibling discovery cannot rely on multicast scouting (macOS never + delivers multicast pinned to lo0), so each worker listens on a + coordinator-allocated loopback port and dials the workers spawned + before it, while coordinator-side sessions dial every live worker. + Spawn order closes the mesh: whatever exists when a session opens is + in its dial list, and everything newer dials it. + """ + worker = PythonWorker() + if self._cfg.transport == "zenoh": + worker.start_process((allocate_mesh_endpoint(), self._mesh_endpoints())) + else: + worker.start_process() + self._workers.append(worker) + if worker.listen_endpoint is not None: + configure_zenoh_mesh(None, self._mesh_endpoints()) + return worker + def start(self) -> None: if self._started: return self._started = True for _ in range(self._n_workers): - worker = PythonWorker() - worker.start_process() - self._workers.append(worker) + self._spawn_worker() logger.info("Worker pool started.", n_workers=self._n_workers) if self._cfg.dtop: @@ -76,9 +98,7 @@ def add_workers(self, n: int) -> None: if not self._started: raise RuntimeError("WorkerManager not started; call start() first") for _ in range(n): - worker = PythonWorker() - worker.start_process() - self._workers.append(worker) + self._spawn_worker() self._n_workers += n logger.info("Added workers to pool.", added=n, total=self._n_workers) @@ -116,9 +136,7 @@ def deploy_fresh( if not self._started: self.start() - worker = PythonWorker() - worker.start_process() - self._workers.append(worker) + worker = self._spawn_worker() self._n_workers += 1 if module_class.dedicated_worker: worker.dedicated = True @@ -146,6 +164,10 @@ def undeploy(self, proxy: ModuleProxyProtocol) -> None: target.shutdown() self._workers.remove(target) self._n_workers = max(0, self._n_workers - 1) + if target.listen_endpoint is not None: + # Future workers and coordinator sessions must not dial the + # departed worker's port; it may be reused by anyone. + configure_zenoh_mesh(None, self._mesh_endpoints()) def deploy_parallel( self, specs: Iterable[ModuleSpec], blueprint_args: Mapping[str, Mapping[str, Any]] @@ -233,6 +255,7 @@ def stop(self) -> None: logger.error(f"Error shutting down worker: {e}", exc_info=True) self._workers.clear() + configure_zenoh_mesh(None, ()) logger.info("All workers shut down") diff --git a/dimos/protocol/service/zenohservice.py b/dimos/protocol/service/zenohservice.py index 22bc364377..1a36566bf8 100644 --- a/dimos/protocol/service/zenohservice.py +++ b/dimos/protocol/service/zenohservice.py @@ -19,11 +19,14 @@ import socket import threading import time -from typing import Any +from typing import TYPE_CHECKING, Any from pydantic import Field import zenoh +if TYPE_CHECKING: + from collections.abc import Sequence + from dimos.protocol.service.spec import BaseConfig, Service from dimos.utils.logging_config import setup_logger @@ -42,9 +45,45 @@ # interface name, and Darwin spells loopback differently. LOOPBACK_INTERFACE = "lo0" if platform.system() == "Darwin" else "lo" +# Explicit loopback endpoints wiring the coordinator and its workers together. +# Sibling discovery cannot rely on multicast scouting: macOS never delivers +# multicast pinned to lo0, so with scouting off the processes would never find +# each other. Instead each worker listens on a coordinator-allocated port and +# dials the workers spawned before it, and coordinator-side sessions dial every +# live worker. Process-wide; consumed by the ZenohConfig default factories. +_mesh_listen: str | None = None +_mesh_connect: tuple[str, ...] = () + + +def allocate_mesh_endpoint() -> str: + """Reserve a free loopback port and return it as a ``tcp/`` locator. + + The probe socket closes before zenoh binds the port at session open; a + collision in that window fails the session open loudly rather than + silently dropping traffic. + """ + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port: int = sock.getsockname()[1] + return f"tcp/127.0.0.1:{port}" + + +def configure_zenoh_mesh(listen: str | None, connect: Sequence[str]) -> None: + """Set the loopback endpoints this process's future zenoh sessions mesh over. + + Workers pass their own ``listen`` endpoint plus the endpoints of the + workers spawned before them; the coordinator passes ``None`` and dials all + live workers. Already-open sessions are not reconfigured: the mesh stays + complete because every session dials whatever existed when it opened and + is dialled by everything newer. + """ + global _mesh_listen, _mesh_connect + _mesh_listen = listen + _mesh_connect = tuple(connect) + def _default_connect_endpoints() -> list[str]: - """Dial known robots directly instead of trusting multicast scouting. + """Dial known robots and mesh siblings instead of trusting multicast scouting. Many APs filter multicast between WiFi clients, so a robot that is perfectly reachable over TCP never answers a scout. When the session is @@ -54,19 +93,23 @@ def _default_connect_endpoints() -> list[str]: """ from dimos.core.global_config import global_config - if global_config.transport != "zenoh": - return [] - ips = [global_config.robot_ip or "", *(global_config.robot_ips or "").split(",")] out: list[str] = [] - for ip in (x.strip() for x in ips): - if not ip: - continue - endpoint = f"tcp/{ip}" if ":" in ip else f"tcp/{ip}:{ROBOT_ZENOH_PORT}" - if endpoint not in out: - out.append(endpoint) + if global_config.transport == "zenoh": + ips = [global_config.robot_ip or "", *(global_config.robot_ips or "").split(",")] + for ip in (x.strip() for x in ips): + if not ip: + continue + endpoint = f"tcp/{ip}" if ":" in ip else f"tcp/{ip}:{ROBOT_ZENOH_PORT}" + if endpoint not in out: + out.append(endpoint) + out.extend(endpoint for endpoint in _mesh_connect if endpoint not in out) return out +def _default_listen_endpoints() -> list[str]: + return [_mesh_listen] if _mesh_listen else [] + + def _default_scouting() -> bool: from dimos.core.global_config import global_config @@ -101,7 +144,7 @@ def endpoint_addresses(endpoint: str) -> set[str]: class ZenohConfig(BaseConfig): mode: str = "peer" connect: list[str] = Field(default_factory=_default_connect_endpoints) - listen: list[str] = [] + listen: list[str] = Field(default_factory=_default_listen_endpoints) # Discover peers across the network. Off keeps discovery on loopback. scouting: bool = Field(default_factory=_default_scouting) # Seconds to block in start() waiting for `connect` endpoints to link. @@ -129,13 +172,28 @@ def acquire(self, config: ZenohConfig) -> zenoh.Session: zconfig.insert_json5("mode", json.dumps(config.mode)) if config.connect: zconfig.insert_json5("connect/endpoints", json.dumps(config.connect)) + # A dial can race a mesh sibling's listener still coming + # up; zenoh's default 1s initial retry would hold back the + # first messages. Retry fast, keep the default backoff cap. + # (Whole object: zenoh rejects inserts at the leaf keys.) + zconfig.insert_json5( + "connect/retry", + json.dumps( + { + "period_init_ms": 100, + "period_max_ms": 4000, + "period_increase_factor": 2, + } + ), + ) if config.listen: zconfig.insert_json5("listen/endpoints", json.dumps(config.listen)) if not config.scouting: - # Loopback multicast stays on so sibling worker processes on - # this host still discover each other -- cutting scouting - # outright leaves them unable to reach one another at all, - # since peers don't route each other's traffic. + # The coordinator and its workers reach each other over the + # explicit mesh endpoints above. Loopback multicast stays on + # for other same-host processes (CLIs attaching to a + # daemon); note macOS never delivers it on lo0, so those + # need network scouting or explicit endpoints there. zconfig.insert_json5( "scouting/multicast/interface", json.dumps(LOOPBACK_INTERFACE) ) @@ -182,7 +240,13 @@ def _await_connect(self, session: zenoh.Session) -> None: Unreachable endpoints are a warning, not an error: one robot being down should not stop the rest of the graph from coming up. """ - pending = {ep: endpoint_addresses(ep) for ep in self.config.connect} + # Mesh endpoints are excluded: a sibling's listener only comes up once + # its own first module deploys, zenoh keeps dialling in the background, + # and the RPC retry loop rides out the gap. Blocking every session + # start on them would stall deploys instead. + pending = { + ep: endpoint_addresses(ep) for ep in self.config.connect if ep not in _mesh_connect + } if not pending or self.config.connect_timeout <= 0: return deadline = time.monotonic() + self.config.connect_timeout