diff --git a/dimos/cli/commands/lifecycle.py b/dimos/cli/commands/lifecycle.py index 114594b0cb..2e40f0f042 100644 --- a/dimos/cli/commands/lifecycle.py +++ b/dimos/cli/commands/lifecycle.py @@ -30,6 +30,7 @@ from dimos.constants import CONFIG_DIR, LOG_DIR from dimos.core.daemon import ( + DAEMON_STATUS_FD_ENV, fork_daemon, install_signal_handlers, read_daemon_status, @@ -222,7 +223,12 @@ def run( logger.info(f"Cleaned {stale} stale run entries") blueprint_name = "-".join(blueprint_names) - run_id = generate_run_id(blueprint_name) + # A macOS daemon re-execs this command. Preserve the launcher's run/log + # identity rather than generating a second one in the fresh interpreter. + inherited_run_id = ( + os.environ.get(DIMOS_RUN_ID_ENV) if daemon and DAEMON_STATUS_FD_ENV in os.environ else None + ) + run_id = inherited_run_id or generate_run_id(blueprint_name) log_dir = LOG_DIR / run_id # Tag every descendant with the run id so the watchdog and stale-run diff --git a/dimos/cli/dimos.py b/dimos/cli/dimos.py index 27e319637b..13596ebf67 100644 --- a/dimos/cli/dimos.py +++ b/dimos/cli/dimos.py @@ -46,6 +46,11 @@ from dotenv import load_dotenv import typer +# Load before CLI imports freeze config/state paths, so a daemon re-exec and +# later status/stop commands resolve the same registry. Keep tests isolated. +if "PYTEST_VERSION" not in os.environ: + load_dotenv() + from dimos.cli.cache import app as cache_app from dimos.cli.can import app as can_app from dimos.cli.cloud import login as cloud_login, logout as cloud_logout, whoami as cloud_whoami @@ -68,7 +73,6 @@ from dimos.cli.hardware_cli import app as hardware_app from dimos.cli.shell import shell from dimos.cli.vqa import app as vqa_app -from dimos.core.global_config import ENV_FILE from dimos.robot.unitree.go2.cli.go2tool import app as go2tool_app main = typer.Typer( @@ -76,9 +80,6 @@ no_args_is_help=True, ) -if ENV_FILE is not None: - load_dotenv() - SIMULATORS = ("mujoco", "dimsim") RECORDERS = ("sqlite", "mcap") diff --git a/dimos/cli/test_dimos.py b/dimos/cli/test_dimos.py index cf8380e6b9..ddf4df7c98 100644 --- a/dimos/cli/test_dimos.py +++ b/dimos/cli/test_dimos.py @@ -303,6 +303,30 @@ def remove(self) -> None: global_config.update(**original_global_config) +@pytest.mark.parametrize( + ("exec_handoff", "expected_run_id"), + [(True, "launcher-run"), (False, "test-alpha")], +) +def test_daemon_reuses_identity_only_for_exec_handoff( + stubbed_run: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, + exec_handoff: bool, + expected_run_id: str, +) -> None: + monkeypatch.setenv("DIMOS_RUN_ID", "launcher-run") + if exec_handoff: + monkeypatch.setenv("_DIMOS_DAEMON_STATUS_FD", "7") + else: + monkeypatch.delenv("_DIMOS_DAEMON_STATUS_FD", raising=False) + monkeypatch.setattr(lifecycle, "fork_daemon", lambda path: (123, 7)) + monkeypatch.setattr(lifecycle, "read_daemon_status", lambda fd: {"ok": True, "n_modules": 1}) + + result = CliRunner().invoke(main, ["run", "alpha", "--daemon"]) + + assert result.exit_code == 0, result.output + assert f"Run ID: {expected_run_id}" in result.output + + def test_run_rejects_record_topics_matching_nothing(stubbed_run: dict[str, Any]) -> None: result = CliRunner().invoke( main, ["--record", "sqlite", "--record-topics", "nope", "run", "alpha"] diff --git a/dimos/core/daemon.py b/dimos/core/daemon.py index 69ea4e865f..f60bfbe817 100644 --- a/dimos/core/daemon.py +++ b/dimos/core/daemon.py @@ -33,6 +33,8 @@ logger = setup_logger() +DAEMON_STATUS_FD_ENV = "_DIMOS_DAEMON_STATUS_FD" + def health_check(coordinator: ModuleCoordinator) -> bool: """Verify all coordinator workers are alive after build. @@ -56,13 +58,32 @@ def fork_daemon(log_dir: Path) -> tuple[int, int]: The grandchild keeps the launcher's stdio so startup output still reaches the terminal; it must call ``redirect_stdio_to_devnull`` and ``write_daemon_status`` once startup succeeds or fails. + + On macOS the grandchild re-execs the CLI before building. Native imports + can initialize CoreFoundation, which is unsafe in a forked interpreter + even if no transport sessions have been opened. """ + inherited_fd = os.environ.pop(DAEMON_STATUS_FD_ENV, None) + if inherited_fd is not None: + write_fd = int(inherited_fd) + # Keep the pipe across this exec only, not later worker/watchdog execs. + os.set_inheritable(write_fd, False) + return 0, write_fd + log_dir.mkdir(parents=True, exist_ok=True) # Anything buffered would otherwise be flushed by both processes. sys.stdout.flush() sys.stderr.flush() read_fd, write_fd = os.pipe() + exec_env = None + exec_argv = None + if sys.platform == "darwin": + os.set_inheritable(write_fd, True) + exec_env = os.environ | {DAEMON_STATUS_FD_ENV: str(write_fd)} + # Preserve interpreter flags and both console-script/python -m entry. + exec_argv = [sys.executable, *sys.orig_argv[1:]] + pid = os.fork() if pid > 0: os.close(write_fd) @@ -75,6 +96,8 @@ def fork_daemon(log_dir: Path) -> tuple[int, int]: # Second fork — can never reacquire a controlling terminal if os.fork() > 0: os._exit(0) + if exec_env is not None and exec_argv is not None: + os.execve(sys.executable, exec_argv, exec_env) return 0, write_fd diff --git a/dimos/core/test_daemon_exec.py b/dimos/core/test_daemon_exec.py new file mode 100644 index 0000000000..b0b0caa4b0 --- /dev/null +++ b/dimos/core/test_daemon_exec.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. + +"""Check the macOS exec handoff without forking the pytest worker.""" + +from contextlib import suppress +import json +import os +from pathlib import Path +import sys +from unittest import mock + +import pytest + +from dimos.core.daemon import fork_daemon, write_daemon_status + + +def test_reexecuted_daemon_consumes_status_pipe_once( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + read_fd, write_fd = os.pipe() + try: + os.set_inheritable(write_fd, True) + monkeypatch.setenv("_DIMOS_DAEMON_STATUS_FD", str(write_fd)) + with mock.patch("os.fork", side_effect=AssertionError("must not fork twice")): + assert fork_daemon(tmp_path) == (0, write_fd) + assert "_DIMOS_DAEMON_STATUS_FD" not in os.environ + assert not os.get_inheritable(write_fd) + write_daemon_status(write_fd, {"ok": True}) + assert json.loads(os.read(read_fd, 100)) == {"ok": True} + finally: + os.close(read_fd) + os.close(write_fd) + + +@pytest.mark.parametrize( + "original_argv", + [ + [ + "python", + "-u", + "-m", + "dimos.cli.dimos", + "--replay", + "run", + "unitree-go2-basic", + "--daemon", + ], + [ + "python", + "-X", + "dev", + "/workspace with spaces/bin/dimos", + "run", + "demo-mcp-stress-test", + "--daemon", + ], + ], +) +def test_macos_grandchild_reexec_preserves_interpreter_and_arguments( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, original_argv: list[str] +) -> None: + monkeypatch.setattr(sys, "platform", "darwin") + monkeypatch.setattr(sys, "orig_argv", original_argv) + monkeypatch.delenv("_DIMOS_DAEMON_STATUS_FD", raising=False) + monkeypatch.setenv("DIMOS_RUN_ID", "launcher-run") + read_fd, write_fd = os.pipe() + try: + with ( + mock.patch("os.pipe", return_value=(read_fd, write_fd)), + mock.patch("os.fork", side_effect=[0, 0]) as fork, + mock.patch("os.setsid") as setsid, + mock.patch("os.execve", side_effect=OSError("exec unavailable")) as execve, + pytest.raises(OSError, match="exec unavailable"), + ): + fork_daemon(tmp_path) + assert fork.call_count == 2 + setsid.assert_called_once_with() + executable, argv, env = execve.call_args.args + assert executable == sys.executable + assert argv == [sys.executable, *original_argv[1:]] + assert env["DIMOS_RUN_ID"] == "launcher-run" + assert env["_DIMOS_DAEMON_STATUS_FD"] == str(write_fd) + assert os.get_inheritable(write_fd) + assert "_DIMOS_DAEMON_STATUS_FD" not in os.environ + finally: + with suppress(OSError): + os.close(read_fd) + os.close(write_fd) + + +def test_linux_grandchild_retains_fork_without_exec( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.delenv("_DIMOS_DAEMON_STATUS_FD", raising=False) + read_fd, write_fd = os.pipe() + try: + with ( + mock.patch("os.pipe", return_value=(read_fd, write_fd)), + mock.patch("os.fork", side_effect=[0, 0]), + mock.patch("os.setsid"), + mock.patch("os.execve") as execve, + ): + assert fork_daemon(tmp_path) == (0, write_fd) + execve.assert_not_called() + assert not os.get_inheritable(write_fd) + finally: + with suppress(OSError): + os.close(read_fd) + os.close(write_fd) diff --git a/dimos/core/test_daemon_zenoh.py b/dimos/core/test_daemon_zenoh.py index 9b6ecf0082..94dac6b34a 100644 --- a/dimos/core/test_daemon_zenoh.py +++ b/dimos/core/test_daemon_zenoh.py @@ -36,6 +36,8 @@ import sys import time +import pytest + _CLIENT = """ from dimos.core.coordination.coordinator_rpc import CoordinatorRPC @@ -60,13 +62,31 @@ def _sweep_leftover_daemons(state_dir: Path) -> None: time.sleep(1.0) -def test_daemon_serves_coordinator_ping_over_zenoh(tmp_path: Path) -> None: +@pytest.mark.parametrize( + "blueprint", + [ + "demo-mcp-stress-test", + # Go2 replay needs go2_short.db, exceeding the standard CI LFS limit. + pytest.param("unitree-go2-basic", marks=pytest.mark.self_hosted), + ], +) +def test_daemon_serves_coordinator_ping_over_zenoh(tmp_path: Path, blueprint: str) -> None: env = os.environ | { "DIMOS_TRANSPORT": "zenoh", # Never touch the developer's real registry/config, even without xdist. "XDG_STATE_HOME": str(tmp_path / "state"), "XDG_CONFIG_HOME": str(tmp_path / "config"), } + cwd = None + if blueprint == "demo-mcp-stress-test": + # Exercise a dotenv-only registry path without reading/writing the + # developer's .env: dotenv searches upward from the imported CLI. + (tmp_path / "dimos").symlink_to(Path(__file__).resolve().parents[1]) + (tmp_path / ".env").write_text(f"XDG_STATE_HOME={env.pop('XDG_STATE_HOME')}\n") + env.pop("PYTEST_VERSION", None) + env["HOME"] = str(tmp_path / "home") + env["PYTHONPATH"] = os.pathsep.join([str(tmp_path), env.get("PYTHONPATH", "")]) + cwd = tmp_path # File, not pipe: the daemon's forkserver inherits the CLI's stdout/stderr # and holds them open forever, so a pipe would block subprocess.run. run_log = tmp_path / "run.log" @@ -77,17 +97,18 @@ def test_daemon_serves_coordinator_ping_over_zenoh(tmp_path: Path) -> None: sys.executable, "-m", "dimos.cli.dimos", + "--replay", "run", - "demo-mcp-stress-test", + blueprint, "--daemon", - "--disable", - "mcp-server", + *(["--disable", "mcp-server"] if blueprint == "demo-mcp-stress-test" else []), "--viewer", "none", "--n-workers", "1", ], env=env, + cwd=cwd, stdin=subprocess.DEVNULL, stdout=log, stderr=log, @@ -98,6 +119,7 @@ def test_daemon_serves_coordinator_ping_over_zenoh(tmp_path: Path) -> None: client = subprocess.run( [sys.executable, "-c", _CLIENT], env=env, + cwd=cwd, capture_output=True, text=True, timeout=60, @@ -106,11 +128,25 @@ def test_daemon_serves_coordinator_ping_over_zenoh(tmp_path: Path) -> None: f"Coordinator/ping unreachable from an out-of-process zenoh client:\n" f"{client.stdout}\n{client.stderr}" ) + status = subprocess.run( + [sys.executable, "-m", "dimos.cli.dimos", "status"], + env=env, + cwd=cwd, + capture_output=True, + text=True, + timeout=30, + ) + assert blueprint in status.stdout, status.stdout + status.stderr + assert list((tmp_path / "state/dimos/runs").glob("*.json")) finally: - subprocess.run( + stopped = subprocess.run( [sys.executable, "-m", "dimos.cli.dimos", "stop"], env=env, + cwd=cwd, capture_output=True, + text=True, timeout=60, ) _sweep_leftover_daemons(tmp_path / "state") + _sweep_leftover_daemons(tmp_path / "home/.local/state") + assert stopped.returncode == 0, stopped.stdout + stopped.stderr