Autoconf non-interactive fix

dimos/protocol/pubsub/impl/test_lcmpubsub.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 from collections.abc import Generator
+import time
 from typing import Any
 
 import pytest
@@ -37,6 +38,7 @@
 def lcm_pub_sub_base() -> Generator[LCMPubSubBase, None, None]:
     lcm = LCMPubSubBase(url=_ISOLATED_LCM_URL)
     lcm.start()
+    time.sleep(0.05)  # let the handler thread enter the LCM loop
     yield lcm
     lcm.stop()
 
@@ -45,6 +47,7 @@ def lcm_pub_sub_base() -> Generator[LCMPubSubBase, None, None]:
 def pickle_lcm() -> Generator[PickleLCM, None, None]:
     lcm = PickleLCM(url=_ISOLATED_LCM_URL)
     lcm.start()
+    time.sleep(0.05)  # let the handler thread enter the LCM loop
     yield lcm
     lcm.stop()
 
@@ -53,6 +56,7 @@ def pickle_lcm() -> Generator[PickleLCM, None, None]:
 def lcm() -> Generator[LCM, None, None]:
     lcm = LCM(url=_ISOLATED_LCM_URL)
     lcm.start()
+    time.sleep(0.05)  # let the handler thread enter the LCM loop
     yield lcm
     lcm.stop()
 

dimos/protocol/service/system_configurator/base.py

@@ -15,32 +15,14 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from functools import cache
 import logging
 import os
 import subprocess
-from typing import Any
 
-import typer
+from dimos.utils import prompt
 
 logger = logging.getLogger(__name__)
 
-# sudo helpers
-
-
-@cache
-def _is_root_user() -> bool:
-    try:
-        return os.geteuid() == 0
-    except AttributeError:
-        return False
-
-
-def sudo_run(*args: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]:
-    if _is_root_user():
-        return subprocess.run(list(args), **kwargs)
-    return subprocess.run(["sudo", *args], **kwargs)
-
 
 def _read_sysctl_int(name: str) -> int | None:
     try:
@@ -63,7 +45,7 @@ def _read_sysctl_int(name: str) -> int | None:
 
 
 def _write_sysctl_int(name: str, value: int) -> None:
-    sudo_run("sysctl", "-w", f"{name}={value}", check=True, text=True, capture_output=False)
+    prompt.sudo_run("sysctl", "-w", f"{name}={value}", check=True, text=True, capture_output=False)
 
 
 # base class for system config checks/requirements
@@ -114,7 +96,7 @@ def configure_system(checks: list[SystemConfigurator], check_only: bool = False)
     if check_only:
         return
 
-    if not typer.confirm("\nApply these changes now?"):
+    if not prompt.confirm("\nApply these changes now?"):
         if any(check.critical for check in failing):
             raise SystemExit(1)
         return

dimos/protocol/service/system_configurator/clock_sync.py

@@ -20,7 +20,8 @@
 import struct
 import time
 
-from dimos.protocol.service.system_configurator.base import SystemConfigurator, sudo_run
+from dimos.protocol.service.system_configurator.base import SystemConfigurator
+from dimos.utils import prompt
 from dimos.utils.human import human_duration
 
 
@@ -122,4 +123,4 @@ def fix(self) -> None:
         # Recompute the corrected time at fix-time (not stale from check-time)
         if cmd[:2] == ["date", "-s"] and self._offset is not None:
             cmd[2] = f"@{time.time() - self._offset:.3f}"
-        sudo_run(*cmd, check=True, text=True, capture_output=True)
+        prompt.sudo_run(*cmd, check=True, text=True, capture_output=True)

dimos/protocol/service/system_configurator/lcm.py

@@ -22,8 +22,8 @@
     SystemConfigurator,
     _read_sysctl_int,
     _write_sysctl_int,
-    sudo_run,
 )
+from dimos.utils import prompt
 
 # specific checks: multicast
 
@@ -126,9 +126,9 @@ def explanation(self) -> str | None:
 
     def fix(self) -> None:
         if not self.loopback_ok:
-            sudo_run(*self.enable_multicast_cmd, check=True, text=True, capture_output=True)
+            prompt.sudo_run(*self.enable_multicast_cmd, check=True, text=True, capture_output=True)
         if not self.route_ok:
-            sudo_run(*self.add_route_cmd, check=True, text=True, capture_output=True)
+            prompt.sudo_run(*self.add_route_cmd, check=True, text=True, capture_output=True)
 
 
 class MulticastConfiguratorMacOS(SystemConfigurator):
@@ -170,7 +170,7 @@ def explanation(self) -> str | None:
     def fix(self) -> None:
         # Delete any existing 224.0.0.0/4 route (e.g. on en0) before adding on lo0,
         # otherwise `route add` fails with "route already in use"
-        sudo_run(
+        prompt.sudo_run(
             "route",
             "delete",
             "-net",
@@ -179,7 +179,7 @@ def fix(self) -> None:
             text=True,
             capture_output=True,
         )
-        sudo_run(*self.add_route_cmd, check=True, text=True, capture_output=True)
+        prompt.sudo_run(*self.add_route_cmd, check=True, text=True, capture_output=True)
 
 
 # specific checks: buffers
@@ -312,7 +312,7 @@ def fix(self) -> None:
         else:
             # Need to raise both soft and hard limits via launchctl
             try:
-                sudo_run(
+                prompt.sudo_run(
                     "launchctl",
                     "limit",
                     "maxfiles",

dimos/protocol/service/test_system_configurator.py

@@ -21,11 +21,9 @@
 
 from dimos.protocol.service.system_configurator.base import (
     SystemConfigurator,
-    _is_root_user,
     _read_sysctl_int,
     _write_sysctl_int,
     configure_system,
-    sudo_run,
 )
 from dimos.protocol.service.system_configurator.clock_sync import ClockSyncConfigurator
 from dimos.protocol.service.system_configurator.lcm import (
@@ -36,43 +34,24 @@
     MulticastConfiguratorLinux,
     MulticastConfiguratorMacOS,
 )
+from dimos.utils import prompt
 
 # Helper function tests
 
 
-class TestIsRootUser:
-    def test_is_root_when_euid_is_zero(self) -> None:
-        # Clear the cache before testing
-        _is_root_user.cache_clear()
-        with patch("os.geteuid", return_value=0):
-            assert _is_root_user() is True
-
-    def test_is_not_root_when_euid_is_nonzero(self) -> None:
-        _is_root_user.cache_clear()
-        with patch("os.geteuid", return_value=1000):
-            assert _is_root_user() is False
-
-    def test_returns_false_when_geteuid_not_available(self) -> None:
-        _is_root_user.cache_clear()
-        with patch("os.geteuid", side_effect=AttributeError):
-            assert _is_root_user() is False
-
-
 class TestSudoRun:
     def test_runs_without_sudo_when_root(self) -> None:
-        _is_root_user.cache_clear()
         with patch("os.geteuid", return_value=0):
             with patch("subprocess.run") as mock_run:
                 mock_run.return_value = MagicMock(returncode=0)
-                sudo_run("echo", "hello", check=True)
+                prompt.sudo_run("echo", "hello", check=True)
                 mock_run.assert_called_once_with(["echo", "hello"], check=True)
 
     def test_runs_with_sudo_when_not_root(self) -> None:
-        _is_root_user.cache_clear()
         with patch("os.geteuid", return_value=1000):
             with patch("subprocess.run") as mock_run:
                 mock_run.return_value = MagicMock(returncode=0)
-                sudo_run("echo", "hello", check=True)
+                prompt.sudo_run("echo", "hello", check=True)
                 mock_run.assert_called_once_with(["sudo", "echo", "hello"], check=True)
 
 
@@ -109,7 +88,6 @@ def test_returns_none_on_exception(self) -> None:
 
 class TestWriteSysctlInt:
     def test_calls_sudo_run_with_correct_args(self) -> None:
-        _is_root_user.cache_clear()
         with patch("os.geteuid", return_value=1000):
             with patch("subprocess.run") as mock_run:
                 mock_run.return_value = MagicMock(returncode=0)
@@ -168,19 +146,19 @@ def test_check_only_mode_does_not_fix(self) -> None:
 
     def test_prompts_user_and_fixes_on_yes(self, mocker) -> None:
         mock_check = MockConfigurator(passes=False)
-        mocker.patch("typer.confirm", return_value=True)
+        mocker.patch("dimos.utils.prompt.confirm", return_value=True)
         configure_system([mock_check])
         assert mock_check.fix_called
 
     def test_does_not_fix_on_no(self, mocker) -> None:
         mock_check = MockConfigurator(passes=False)
-        mocker.patch("typer.confirm", return_value=False)
+        mocker.patch("dimos.utils.prompt.confirm", return_value=False)
         configure_system([mock_check])
         assert not mock_check.fix_called
 
     def test_exits_on_no_with_critical_check(self, mocker) -> None:
         mock_check = MockConfigurator(passes=False, is_critical=True)
-        mocker.patch("typer.confirm", return_value=False)
+        mocker.patch("dimos.utils.prompt.confirm", return_value=False)
         with pytest.raises(SystemExit) as exc_info:
             configure_system([mock_check])
         assert exc_info.value.code == 1
@@ -248,7 +226,6 @@ def test_explanation_includes_needed_commands(self) -> None:
         assert "ip route add 224.0.0.0/4 dev lo" in explanation
 
     def test_fix_runs_needed_commands(self) -> None:
-        _is_root_user.cache_clear()
         configurator = MulticastConfiguratorLinux()
         configurator.loopback_ok = False
         configurator.route_ok = False
@@ -292,7 +269,6 @@ def test_explanation_includes_route_command(self) -> None:
         assert "route add -net 224.0.0.0/4 -interface lo0" in explanation
 
     def test_fix_runs_route_command(self) -> None:
-        _is_root_user.cache_clear()
         configurator = MulticastConfiguratorMacOS()
         with patch("os.geteuid", return_value=0):
             with patch("subprocess.run") as mock_run:
@@ -464,7 +440,6 @@ def test_fix_does_nothing_when_already_sufficient(self) -> None:
             mock_setrlimit.assert_not_called()
 
     def test_fix_uses_launchctl_when_hard_limit_low(self) -> None:
-        _is_root_user.cache_clear()
         configurator = MaxFileConfiguratorMacOS(target=65536)
         configurator.current_soft = 256
         configurator.current_hard = 10240
@@ -594,7 +569,6 @@ def test_explanation_returns_none_when_ntp_unreachable(self) -> None:
         assert configurator.explanation() is None
 
     def test_fix_on_linux_with_ntpdate(self) -> None:
-        _is_root_user.cache_clear()
         configurator = ClockSyncConfigurator()
         with (
             patch(
@@ -615,7 +589,6 @@ def test_fix_on_linux_with_ntpdate(self) -> None:
             assert "ntpdate" in mock_run.call_args_list[0][0][0]
 
     def test_fix_on_linux_sntp_fallback(self) -> None:
-        _is_root_user.cache_clear()
         configurator = ClockSyncConfigurator()
         with (
             patch(
@@ -636,7 +609,6 @@ def test_fix_on_linux_sntp_fallback(self) -> None:
             assert "sntp" in mock_run.call_args_list[0][0][0]
 
     def test_fix_on_linux_date_fallback(self) -> None:
-        _is_root_user.cache_clear()
         configurator = ClockSyncConfigurator()
         configurator._offset = 1.0
         with (
@@ -658,7 +630,6 @@ def test_fix_on_linux_date_fallback(self) -> None:
             assert "date" in mock_run.call_args_list[0][0][0]
 
     def test_fix_on_macos(self) -> None:
-        _is_root_user.cache_clear()
         configurator = ClockSyncConfigurator()
         with (
             patch(

dimos/utils/prompt.py

@@ -0,0 +1,53 @@
+# Copyright 2025-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.
+
+"""
+Prompts that are safe to call in modules (despite potentially no stdin, or potentially a TUI that eats/controls the stdin)
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from typing import Any
+
+
+def confirm(message: str, *, default: bool = True) -> bool:
+    """Ask yes/no.
+
+    In non-interactive mode (no tty), returns *default* without prompting
+    — useful for daemons and CI where stdin is unavailable.
+
+    In interactive mode, no default is pre-selected so the user must
+    explicitly type ``y`` or ``n``.  This prevents accidental Enter-mashing
+    from silently triggering system changes (some of which require sudo).
+    """
+    if not sys.stdin.isatty():
+        return default
+    import typer
+
+    # No default in interactive mode — require explicit y/n
+    return typer.confirm(message)
+
+
+def sudo_run(*args: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]:
+    """Run a command, prepending sudo if not already root."""
+    try:
+        is_root = os.geteuid() == 0
+    except AttributeError:
+        is_root = False
+    if is_root:
+        return subprocess.run(list(args), **kwargs)
+    return subprocess.run(["sudo", *args], **kwargs)