Fix leaks

.gitignore

@@ -43,5 +43,3 @@ data/*
 !data/.lfs/
 FastSAM-x.pt
 yolo11n.pt
-
-/thread_monitor_report.csv

dimos/agents2/skills/conftest.py

@@ -15,6 +15,7 @@
 import pytest
 import reactivex as rx
 from functools import partial
+from reactivex.scheduler import ThreadPoolScheduler
 
 from dimos.agents2.skills.gps_nav_skill import GpsNavSkillContainer
 from dimos.agents2.skills.navigation import NavigationSkillContainer
@@ -26,6 +27,21 @@
 from dimos.msgs.sensor_msgs import Image
 
 
+@pytest.fixture(autouse=True)
+def cleanup_threadpool_scheduler(monkeypatch):
+    # TODO: get rid of this global threadpool
+    """Clean up and recreate the global ThreadPoolScheduler after each test."""
+    # Disable ChromaDB telemetry to avoid leaking threads
+    monkeypatch.setenv("CHROMA_ANONYMIZED_TELEMETRY", "False")
+    yield
+    from dimos.utils import threadpool
+
+    # Shutdown the global scheduler's executor
+    threadpool.scheduler.executor.shutdown(wait=True)
+    # Recreate it for the next test
+    threadpool.scheduler = ThreadPoolScheduler(max_workers=threadpool.get_max_workers())
+
+
 @pytest.fixture
 def fake_robot(mocker):
     return mocker.MagicMock()

dimos/agents2/skills/google_maps_skill_container.py

@@ -52,6 +52,7 @@ def __enter__(self) -> "GoogleMapsSkillContainer":
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self._disposables.dispose()
+        self.stop()
         return False
 
     def _on_gps_location(self, location: LatLon) -> None:

dimos/agents2/skills/gps_nav_skill.py

@@ -56,6 +56,7 @@ def __enter__(self) -> "GpsNavSkillContainer":
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self._disposables.dispose()
+        self.stop()
         return False
 
     def _on_gps_location(self, location: LatLon) -> None:

dimos/agents2/skills/navigation.py

@@ -17,11 +17,13 @@
 import cv2
 from reactivex import Observable
 
+from dimos.models.vl.qwen import QwenVlModel
 from dimos.msgs.sensor_msgs import Image
+from dimos.navigation.visual.query import get_object_bbox_from_image
 from dimos.protocol.skill.skill import SkillContainer, skill
 from dimos.robot.robot import UnitreeRobot
 from dimos.types.robot_location import RobotLocation
-from dimos.models.qwen.video_query import BBox, get_bbox_from_qwen_frame
+from dimos.models.qwen.video_query import BBox
 from dimos.msgs.geometry_msgs import PoseStamped
 from dimos.msgs.geometry_msgs.Vector3 import make_vector3
 from dimos.utils.transform_utils import euler_to_quaternion, quaternion_to_euler
@@ -46,6 +48,7 @@ def __init__(self, robot: UnitreeRobot, video_stream: Observable[Image]):
         self._video_stream = video_stream
         self._similarity_threshold = 0.23
         self._started = False
+        self._vl_model = QwenVlModel()
 
     def __enter__(self) -> "NavigationSkillContainer":
         unsub = self._video_stream.subscribe(self._on_video)
@@ -55,6 +58,7 @@ def __enter__(self) -> "NavigationSkillContainer":
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self._disposables.dispose()
+        self.stop()
         return False
 
     def _on_video(self, image: Image) -> None:
@@ -174,7 +178,7 @@ def _get_bbox_for_current_frame(self, query: str) -> Optional[BBox]:
         if frame is None:
             return None
 
-        return get_bbox_from_qwen_frame(frame, object_name=query)
+        return get_object_bbox_from_image(self._vl_model, frame, query)
 
     def _navigate_using_semantic_map(self, query: str) -> str:
         results = self._robot.spatial_memory.query_by_text(query)

dimos/agents2/skills/osm.py

@@ -52,6 +52,7 @@ def __enter__(self) -> "OsmSkillContainer":
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self._disposables.dispose()
+        self.stop()
         return False
 
     def _on_gps_location(self, location: LatLon) -> None:

dimos/agents2/temp/test_unitree_agent_query.py

@@ -101,6 +101,9 @@ def test_sync_query_with_thread():
     agent.register_skills(container)
     agent.start()
 
+    # Track the thread we might create
+    loop_thread = None
+
     # The agent's event loop should be running in the Module's thread
     # Let's check if it's running
     if agent._loop and agent._loop.is_running():
@@ -113,8 +116,8 @@ def run_loop():
             asyncio.set_event_loop(agent._loop)
             agent._loop.run_forever()
 
-        thread = threading.Thread(target=run_loop, daemon=True)
-        thread.start()
+        loop_thread = threading.Thread(target=run_loop, daemon=False, name="EventLoopThread")
+        loop_thread.start()
         time.sleep(1)  # Give loop time to start
         logger.info("Started event loop in thread")
 
@@ -129,9 +132,24 @@ def run_loop():
 
         traceback.print_exc()
 
-    # Clean up
+    # Clean up properly
+    # First stop the agent (this should stop its internal loop if any)
     agent.stop()
 
+    # Then stop the manually created event loop thread if we created one
+    if loop_thread and loop_thread.is_alive():
+        logger.info("Stopping manually created event loop thread...")
+        # Stop the event loop
+        if agent._loop and agent._loop.is_running():
+            agent._loop.call_soon_threadsafe(agent._loop.stop)
+        # Wait for thread to finish
+        loop_thread.join(timeout=5)
+        if loop_thread.is_alive():
+            logger.warning("Thread did not stop cleanly within timeout")
+
+    # Finally close the container
+    container._close_module()
+
 
 # def test_with_real_module_system():
 #     """Test using the real DimOS module system (like in test_agent.py)."""

dimos/agents2/temp/test_unitree_skill_container.py

@@ -19,6 +19,7 @@
 """
 
 import sys
+import time
 from pathlib import Path
 
 # Add parent directories to path
@@ -39,19 +40,25 @@ def test_skill_container_creation():
     # Create container without robot (for testing)
     container = UnitreeSkillContainer(robot=None)
 
-    # Get available skills from the container
-    skills = container.skills()
-
-    print(f"Number of skills registered: {len(skills)}")
-    print("\nAvailable skills:")
-    for name, skill_config in list(skills.items())[:10]:  # Show first 10
-        print(
-            f"  - {name}: {skill_config.description if hasattr(skill_config, 'description') else 'No description'}"
-        )
-    if len(skills) > 10:
-        print(f"  ... and {len(skills) - 10} more skills")
-
-    return container, skills
+    try:
+        # Get available skills from the container
+        skills = container.skills()
+
+        print(f"Number of skills registered: {len(skills)}")
+        print("\nAvailable skills:")
+        for name, skill_config in list(skills.items())[:10]:  # Show first 10
+            print(
+                f"  - {name}: {skill_config.description if hasattr(skill_config, 'description') else 'No description'}"
+            )
+        if len(skills) > 10:
+            print(f"  ... and {len(skills) - 10} more skills")
+
+        return container, skills
+    finally:
+        # Ensure proper cleanup
+        container._close_module()
+        # Small delay to allow threads to finish cleanup
+        time.sleep(0.1)
 
 
 def test_agent_with_skills():
@@ -60,80 +67,60 @@ def test_agent_with_skills():
 
     # Create skill container
     container = UnitreeSkillContainer(robot=None)
+    agent = None
 
-    # Create agent with configuration passed directly
-    agent = Agent(
-        system_prompt="You are a helpful robot assistant that can control a Unitree Go2 robot.",
-        model=Model.GPT_4O_MINI,
-        provider=Provider.OPENAI,
-    )
+    try:
+        # Create agent with configuration passed directly
+        agent = Agent(
+            system_prompt="You are a helpful robot assistant that can control a Unitree Go2 robot.",
+            model=Model.GPT_4O_MINI,
+            provider=Provider.OPENAI,
+        )
 
-    # Register skills
-    agent.register_skills(container)
+        # Register skills
+        agent.register_skills(container)
 
-    print("Agent created and skills registered successfully!")
+        print("Agent created and skills registered successfully!")
 
-    # Get tools to verify
-    tools = agent.get_tools()
-    print(f"Agent has access to {len(tools)} tools")
+        # Get tools to verify
+        tools = agent.get_tools()
+        print(f"Agent has access to {len(tools)} tools")
 
-    return agent
+        return agent
+    finally:
+        # Ensure proper cleanup in order
+        if agent:
+            agent.stop()
+        container._close_module()
+        # Small delay to allow threads to finish cleanup
+        time.sleep(0.1)
 
 
 def test_skill_schemas():
     """Test that skill schemas are properly generated for LangChain."""
     print("\n=== Testing Skill Schemas ===")
 
     container = UnitreeSkillContainer(robot=None)
-    skills = container.skills()
-
-    # Check a few key skills (using snake_case names now)
-    skill_names = ["move", "wait", "stand_up", "sit", "front_flip", "dance1"]
-
-    for name in skill_names:
-        if name in skills:
-            skill_config = skills[name]
-            print(f"\n{name} skill:")
-            print(f"  Config: {skill_config}")
-            if hasattr(skill_config, "schema"):
-                print(
-                    f"  Schema keys: {skill_config.schema.keys() if skill_config.schema else 'None'}"
-                )
-        else:
-            print(f"\nWARNING: Skill '{name}' not found!")
-
-
-def main():
-    """Run all tests."""
-    print("=" * 60)
-    print("Testing UnitreeSkillContainer with agents2 Framework")
-    print("=" * 60)
 
     try:
-        # Test 1: Container creation
-        container, skills = test_skill_container_creation()
-
-        # Test 2: Agent with skills
-        agent = test_agent_with_skills()
-
-        # Test 3: Skill schemas
-        test_skill_schemas()
-
-        # Test 4: Simple query (async)
-        # asyncio.run(test_simple_query())
-        print("\n=== Async query test skipped (would require running agent) ===")
-
-        print("\n" + "=" * 60)
-        print("All tests completed successfully!")
-        print("=" * 60)
-
-    except Exception as e:
-        print(f"\nERROR during testing: {e}")
-        import traceback
-
-        traceback.print_exc()
-        sys.exit(1)
-
-
-if __name__ == "__main__":
-    main()
+        skills = container.skills()
+
+        # Check a few key skills (using snake_case names now)
+        skill_names = ["move", "wait", "stand_up", "sit", "front_flip", "dance1"]
+
+        for name in skill_names:
+            if name in skills:
+                skill_config = skills[name]
+                print(f"\n{name} skill:")
+                print(f"  Config: {skill_config}")
+                if hasattr(skill_config, "schema"):
+                    print(
+                        f"  Schema keys: {skill_config.schema.keys() if skill_config.schema else 'None'}"
+                    )
+            else:
+                print(f"\nWARNING: Skill '{name}' not found!")
+    finally:
+        # Ensure proper cleanup
+        container._close_module()
+        # Small delay to allow threads to finish cleanup
+        time.sleep(0.1)

dimos/agents2/test_mock_agent.py

@@ -128,6 +128,8 @@ def test_image_tool_call():
     ]
     assert len(human_messages_with_images) >= 0  # May have image messages
     agent.stop()
+    test_skill_module.stop()
+    dimos.close_all()
 
 
 @pytest.mark.tool