Mac CI fixes

.github/workflows/ci.yml

@@ -644,11 +644,10 @@ jobs:
                 - /var/cache/dimos-root-cache:/root/.cache
             markers: "self_hosted or skipif_no_ros"
             experimental: false
-          # macOS runner disabled for now — we don't want Mac tests.
-          # - os: macOS
-          #   container: null  # run on host — `container:` is Linux-only
-          #   markers: "self_hosted"
-          #   experimental: true
+          - os: macOS
+            container: null  # run on host — `container:` is Linux-only
+            markers: "self_hosted"
+            experimental: false
     runs-on:
       - self-hosted
       - ${{ matrix.os }}

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:

dimos/core/coordination/worker_manager_python.py

@@ -21,6 +21,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
 
@@ -41,14 +42,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:
@@ -64,9 +86,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)
 
@@ -104,9 +124,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
@@ -134,6 +152,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]) -> list[ModuleProxyProtocol]:
         if self._closed:
@@ -213,6 +235,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")
 

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