refactor: defer AsyncModuleThread loop creation to start()

AsyncModuleThread no longer spawns the event loop thread in __init__.
The loop is created on the first call to start(), which ModuleBase.start()
now calls. This means module construction no longer has side effects —
no threads are spawned until the module is explicitly started.

Author: jeff-hykin

dimos/core/module.py

@@ -123,6 +123,7 @@ def start(self) -> None:
             if state == "stopped":
                 raise RuntimeError(f"{type(self).__name__} cannot be restarted after stop")
             self.mod_state.set("started")
+        self._async_thread.start()
 
     @rpc
     def stop(self) -> None:

dimos/utils/test_thread_utils.py

@@ -339,9 +339,10 @@ def stop_it() -> None:
     def test_close_timeout_respected(self) -> None:
         """If the thread ignores the stop signal, stop() should return after close_timeout."""
         mod = FakeModule()
+        cancel = threading.Event()
 
         def stubborn_target() -> None:
-            time.sleep(10)  # ignores stopping signal
+            cancel.wait(10)  # blocks but can be released for cleanup
 
         mt = ModuleThread(
             module=mod, target=stubborn_target, name="test-timeout", close_timeout=0.2
@@ -351,6 +352,9 @@ def stubborn_target() -> None:
         mt.stop()
         elapsed = time.monotonic() - start
         assert elapsed < 1.0, f"stop() took {elapsed}s, expected ~0.2s"
+        # Release the thread so it doesn't leak
+        cancel.set()
+        mt._thread.join(timeout=1.0)
 
     def test_stop_concurrent_with_dispose(self) -> None:
         """Calling stop() and dispose() concurrently should not crash."""
@@ -382,6 +386,7 @@ class TestAsyncModuleThread:
     def test_creates_loop_and_thread(self) -> None:
         mod = FakeModule()
         amt = AsyncModuleThread(module=mod)
+        amt.start()
         assert amt.loop is not None
         assert amt.loop.is_running()
         assert amt.is_alive
@@ -391,13 +396,15 @@ def test_creates_loop_and_thread(self) -> None:
     def test_stop_idempotent(self) -> None:
         mod = FakeModule()
         amt = AsyncModuleThread(module=mod)
+        amt.start()
         amt.stop()
         amt.stop()  # should not raise
         amt.stop()
 
     def test_dispose_stops_loop(self) -> None:
         mod = FakeModule()
         amt = AsyncModuleThread(module=mod)
+        amt.start()
         assert amt.is_alive
         mod.dispose()
         time.sleep(0.1)
@@ -406,6 +413,7 @@ def test_dispose_stops_loop(self) -> None:
     def test_can_schedule_coroutine(self) -> None:
         mod = FakeModule()
         amt = AsyncModuleThread(module=mod)
+        amt.start()
         result = []
 
         async def coro() -> None:
@@ -420,6 +428,7 @@ def test_stop_with_pending_work(self) -> None:
         """Stop should succeed even with long-running tasks on the loop."""
         mod = FakeModule()
         amt = AsyncModuleThread(module=mod)
+        amt.start()
         started = threading.Event()
 
         async def slow_coro() -> None:
@@ -437,6 +446,7 @@ async def slow_coro() -> None:
     def test_concurrent_stop(self) -> None:
         mod = FakeModule()
         amt = AsyncModuleThread(module=mod)
+        amt.start()
         errors = []
 
         def stop_it() -> None:

dimos/utils/thread_utils.py

@@ -201,6 +201,18 @@ def __init__(
         self._stopped = ThreadSafeVal(False)
         self._owns_loop = False
         self._thread: threading.Thread | None = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._module_name = type(module).__name__
+
+        module._disposables.add(Disposable(self.stop))
+
+    def start(self) -> None:
+        """Create (or reuse) the event loop and start the driver thread.
+
+        Safe to call multiple times — subsequent calls are no-ops.
+        """
+        if self._loop is not None:
+            return
 
         try:
             self._loop = asyncio.get_running_loop()
@@ -211,12 +223,10 @@ def __init__(
             self._thread = threading.Thread(
                 target=self._loop.run_forever,
                 daemon=True,
-                name=f"{type(module).__name__}-event-loop",
+                name=f"{self._module_name}-event-loop",
             )
             self._thread.start()
 
-        module._disposables.add(Disposable(self.stop))
-
     @property
     def loop(self) -> asyncio.AbstractEventLoop:
         """The managed event loop."""
@@ -237,7 +247,7 @@ def stop(self) -> None:
                 return
             self._stopped.set(True)
 
-        if self._owns_loop and self._loop.is_running():
+        if self._owns_loop and self._loop is not None and self._loop.is_running():
             self._loop.call_soon_threadsafe(self._loop.stop)
 
         if self._thread is not None and self._thread.is_alive():