diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml
index 03de5c3d15..361ef66bf8 100644
--- a/.github/workflows/docker.yml
+++ b/.github/workflows/docker.yml
@@ -212,70 +212,9 @@ jobs:
     uses: ./.github/workflows/tests.yml
     secrets: inherit
     with:
-      cmd: "pytest && pytest -m ros" # run tests that depend on ros as well
+      cmd: "pytest --durations=0 -m 'not (tool or mujoco)'"
       dev-image: ros-dev:${{ (needs.check-changes.outputs.python == 'true' || needs.check-changes.outputs.dev == 'true' || needs.check-changes.outputs.ros == 'true') && needs.ros-dev.result == 'success' && needs.check-changes.outputs.branch-tag || 'dev' }}
 
-  run-tests:
-    needs: [check-changes, dev]
-    if: ${{
-      always() &&
-      needs.check-changes.result == 'success' &&
-      (needs.check-changes.outputs.tests == 'true' ||
-       needs.check-changes.outputs.python == 'true' ||
-       needs.check-changes.outputs.dev == 'true')
-      }}
-    uses: ./.github/workflows/tests.yml
-    secrets: inherit
-    with:
-      cmd: "pytest"
-      dev-image: dev:${{ (needs.check-changes.outputs.python == 'true' || needs.check-changes.outputs.dev == 'true') && needs.dev.result == 'success' && needs.check-changes.outputs.branch-tag || 'dev' }}
-
-  # we run in parallel with normal tests for speed
-  run-heavy-tests:
-    needs: [check-changes, dev]
-    if: ${{
-      always() &&
-      needs.check-changes.result == 'success' &&
-      (needs.check-changes.outputs.tests == 'true' ||
-       needs.check-changes.outputs.python == 'true' ||
-       needs.check-changes.outputs.dev == 'true')
-      }}
-    uses: ./.github/workflows/tests.yml
-    secrets: inherit
-    with:
-      cmd: "pytest -m heavy"
-      dev-image: dev:${{ (needs.check-changes.outputs.python == 'true' || needs.check-changes.outputs.dev == 'true') && needs.dev.result == 'success' && needs.check-changes.outputs.branch-tag || 'dev' }}
-
-  run-lcm-tests:
-    needs: [check-changes, dev]
-    if: ${{
-      always() &&
-      needs.check-changes.result == 'success' &&
-      (needs.check-changes.outputs.tests == 'true' ||
-       needs.check-changes.outputs.python == 'true' ||
-       needs.check-changes.outputs.dev == 'true')
-      }}
-    uses: ./.github/workflows/tests.yml
-    secrets: inherit
-    with:
-      cmd: "pytest -m lcm"
-      dev-image: dev:${{ (needs.check-changes.outputs.python == 'true' || needs.check-changes.outputs.dev == 'true') && needs.dev.result == 'success' && needs.check-changes.outputs.branch-tag || 'dev' }}
-
-  run-integration-tests:
-    needs: [check-changes, dev]
-    if: ${{
-      always() &&
-      needs.check-changes.result == 'success' &&
-      (needs.check-changes.outputs.tests == 'true' ||
-       needs.check-changes.outputs.python == 'true' ||
-       needs.check-changes.outputs.dev == 'true')
-      }}
-    uses: ./.github/workflows/tests.yml
-    secrets: inherit
-    with:
-      cmd: "pytest -m integration"
-      dev-image: dev:${{ (needs.check-changes.outputs.python == 'true' || needs.check-changes.outputs.dev == 'true') && needs.dev.result == 'success' && needs.check-changes.outputs.branch-tag || 'dev' }}
-
   run-mypy:
     needs: [check-changes, ros-dev]
     if: ${{
@@ -292,43 +231,8 @@ jobs:
       cmd: "MYPYPATH=/opt/ros/humble/lib/python3.10/site-packages mypy dimos"
       dev-image: ros-dev:${{ (needs.check-changes.outputs.python == 'true' || needs.check-changes.outputs.dev == 'true' || needs.check-changes.outputs.ros == 'true') && needs.ros-dev.result == 'success' && needs.check-changes.outputs.branch-tag || 'dev' }}
 
-  # Run module tests directly to avoid pytest forking issues
-  # run-module-tests:
-  #   needs: [check-changes, dev]
-  #   if: ${{
-  #     always() &&
-  #     needs.check-changes.result == 'success' &&
-  #     ((needs.dev.result == 'success') ||
-  #     (needs.dev.result == 'skipped' &&
-  #     needs.check-changes.outputs.tests == 'true'))
-  #     }}
-  #   runs-on: [self-hosted, x64, 16gb]
-  #   container:
-  #     image: ghcr.io/dimensionalos/dev:${{ needs.check-changes.outputs.dev == 'true' && needs.dev.result == 'success' && needs.check-changes.outputs.branch-tag || 'dev' }}
-  #   steps:
-  #     - name: Fix permissions
-  #       run: |
-  #         sudo chown -R $USER:$USER ${{ github.workspace }} || true
-  #
-  #     - uses: actions/checkout@v4
-  #       with:
-  #         lfs: true
-  #
-  #     - name: Configure Git LFS
-  #       run: |
-  #         git config --global --add safe.directory '*'
-  #         git lfs install
-  #         git lfs fetch
-  #         git lfs checkout
-  #
-  #     - name: Run module tests
-  #       env:
-  #         CI: "true"
-  #       run: |
-  #         /entrypoint.sh bash -c "pytest -m module"
-
   ci-complete:
-    needs: [check-changes, ros, python, ros-python, dev, ros-dev, run-tests, run-heavy-tests, run-lcm-tests, run-integration-tests, run-ros-tests, run-mypy]
+    needs: [check-changes, ros, python, ros-python, dev, ros-dev, run-ros-tests, run-mypy]
     runs-on: [self-hosted, Linux]
     if: always()
     steps:
diff --git a/bin/pytest-slow b/bin/pytest-slow
index 85643d4413..9f9d5ae611 100755
--- a/bin/pytest-slow
+++ b/bin/pytest-slow
@@ -3,4 +3,4 @@
 set -euo pipefail
 
 . .venv/bin/activate
-exec pytest "$@" -m 'not (tool or module or neverending or mujoco)' dimos
+exec pytest "$@" -m 'not (tool or mujoco)' dimos
diff --git a/dimos/agents/mcp/test_mcp_client.py b/dimos/agents/mcp/test_mcp_client.py
index be4a09d5b9..946bdc4eb8 100644
--- a/dimos/agents/mcp/test_mcp_client.py
+++ b/dimos/agents/mcp/test_mcp_client.py
@@ -29,7 +29,7 @@ def add(self, x: int, y: int) -> str:
         return str(x + y)
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 @pytest.mark.parametrize("dask", [False, True])
 def test_can_call_tool(dask, agent_setup):
     history = agent_setup(
@@ -66,7 +66,7 @@ def register_user(self, name: str) -> str:
         return "User name registered successfully."
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 @pytest.mark.parametrize("dask", [False, True])
 def test_can_call_again_on_error(dask, agent_setup):
     history = agent_setup(
@@ -118,7 +118,7 @@ def go_to_location(self, description: str) -> str:
         return f"Going to the {description}."
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_multiple_tool_calls_with_multiple_messages(agent_setup):
     history = agent_setup(
         blueprints=[MultipleTools.blueprint(), NavigationSkill.blueprint()],
@@ -172,7 +172,7 @@ def test_multiple_tool_calls_with_multiple_messages(agent_setup):
     assert len(go_to_location_calls) == 2
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_prompt(agent_setup):
     history = agent_setup(
         blueprints=[],
@@ -190,7 +190,7 @@ def take_a_picture(self) -> Image:
         return Image.from_file(get_data("cafe-smol.jpg")).to_rgb()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_image(agent_setup):
     history = agent_setup(
         blueprints=[Visualizer.blueprint()],
diff --git a/dimos/agents/skills/test_google_maps_skill_container.py b/dimos/agents/skills/test_google_maps_skill_container.py
index 84da91e886..1d8e4549b0 100644
--- a/dimos/agents/skills/test_google_maps_skill_container.py
+++ b/dimos/agents/skills/test_google_maps_skill_container.py
@@ -70,7 +70,7 @@ def __init__(self):
         self._max_valid_distance = 20000
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_where_am_i(agent_setup) -> None:
     history = agent_setup(
         blueprints=[FakeGPS.blueprint(), MockedWhereAmISkill.blueprint()],
@@ -80,7 +80,7 @@ def test_where_am_i(agent_setup) -> None:
     assert "bourbon" in history[-1].content.lower()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_get_gps_position_for_queries(agent_setup) -> None:
     history = agent_setup(
         blueprints=[FakeGPS.blueprint(), MockedPositionSkill.blueprint()],
diff --git a/dimos/agents/skills/test_gps_nav_skills.py b/dimos/agents/skills/test_gps_nav_skills.py
index afcb4d36d0..d701d469ca 100644
--- a/dimos/agents/skills/test_gps_nav_skills.py
+++ b/dimos/agents/skills/test_gps_nav_skills.py
@@ -35,7 +35,7 @@ def __init__(self):
         self._max_valid_distance = 50000
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_set_gps_travel_points(agent_setup) -> None:
     history = agent_setup(
         blueprints=[FakeGPS.blueprint(), MockedGpsNavSkill.blueprint()],
@@ -50,7 +50,7 @@ def test_set_gps_travel_points(agent_setup) -> None:
     assert "success" in history[-1].content.lower()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_set_gps_travel_points_multiple(agent_setup) -> None:
     history = agent_setup(
         blueprints=[FakeGPS.blueprint(), MockedGpsNavSkill.blueprint()],
diff --git a/dimos/agents/skills/test_navigation.py b/dimos/agents/skills/test_navigation.py
index 91737ada77..a7505b23c7 100644
--- a/dimos/agents/skills/test_navigation.py
+++ b/dimos/agents/skills/test_navigation.py
@@ -72,7 +72,7 @@ def _navigate_using_semantic_map(self, query):
         return f"Successfuly arrived at '{query}'"
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_stop_movement(agent_setup) -> None:
     history = agent_setup(
         blueprints=[
@@ -86,7 +86,7 @@ def test_stop_movement(agent_setup) -> None:
     assert "stopped" in history[-1].content.lower()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_start_exploration(agent_setup) -> None:
     history = agent_setup(
         blueprints=[
@@ -102,7 +102,7 @@ def test_start_exploration(agent_setup) -> None:
     assert "explor" in history[-1].content.lower()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_go_to_semantic_location(agent_setup) -> None:
     history = agent_setup(
         blueprints=[
diff --git a/dimos/agents/skills/test_unitree_skill_container.py b/dimos/agents/skills/test_unitree_skill_container.py
index ea1cfba5cf..dde7239bbd 100644
--- a/dimos/agents/skills/test_unitree_skill_container.py
+++ b/dimos/agents/skills/test_unitree_skill_container.py
@@ -29,7 +29,7 @@ def __init__(self):
         self._bound_rpc_calls["GO2Connection.publish_request"] = lambda *args, **kwargs: None
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_pounce(agent_setup) -> None:
     history = agent_setup(
         blueprints=[MockedUnitreeSkill.blueprint()],
diff --git a/dimos/agents/test_agent.py b/dimos/agents/test_agent.py
index da69dfb7dc..cd571a56ae 100644
--- a/dimos/agents/test_agent.py
+++ b/dimos/agents/test_agent.py
@@ -29,7 +29,7 @@ def add(self, x: int, y: int) -> str:
         return str(x + y)
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 @pytest.mark.parametrize("dask", [False, True])
 def test_can_call_tool(dask, agent_setup):
     history = agent_setup(
@@ -68,7 +68,7 @@ def register_user(self, name: str) -> str:
         return "User name registered successfully."
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 @pytest.mark.parametrize("dask", [False, True])
 def test_can_call_again_on_error(dask, agent_setup):
     history = agent_setup(
@@ -120,7 +120,7 @@ def go_to_location(self, description: str) -> str:
         return f"Going to the {description}."
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_multiple_tool_calls_with_multiple_messages(agent_setup):
     history = agent_setup(
         blueprints=[MultipleTools.blueprint(), NavigationSkill.blueprint()],
@@ -174,7 +174,7 @@ def test_multiple_tool_calls_with_multiple_messages(agent_setup):
     assert len(go_to_location_calls) == 2
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_prompt(agent_setup):
     history = agent_setup(
         blueprints=[],
@@ -192,7 +192,7 @@ def take_a_picture(self) -> Image:
         return Image.from_file(get_data("cafe-smol.jpg")).to_rgb()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_image(agent_setup):
     history = agent_setup(
         blueprints=[Visualizer.blueprint()],
diff --git a/dimos/agents_deprecated/memory/test_image_embedding.py b/dimos/agents_deprecated/memory/test_image_embedding.py
index 89f0716e7e..fd8cef696e 100644
--- a/dimos/agents_deprecated/memory/test_image_embedding.py
+++ b/dimos/agents_deprecated/memory/test_image_embedding.py
@@ -22,12 +22,13 @@
 import numpy as np
 import pytest
 from reactivex import operators as ops
+from reactivex.scheduler import ThreadPoolScheduler
 
 from dimos.agents_deprecated.memory.image_embedding import ImageEmbeddingProvider
 from dimos.stream.video_provider import VideoProvider
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 class TestImageEmbedding:
     """Test class for CLIP image embedding functionality."""
 
@@ -45,6 +46,7 @@ def test_clip_embedding_initialization(self) -> None:
 
     def test_clip_embedding_process_video(self) -> None:
         """Test CLIP embedding provider can process video frames and return embeddings."""
+        test_scheduler = ThreadPoolScheduler(max_workers=4)
         try:
             from dimos.utils.data import get_data
 
@@ -53,7 +55,9 @@ def test_clip_embedding_process_video(self) -> None:
             embedding_provider = ImageEmbeddingProvider(model_name="clip", dimensions=512)
 
             assert os.path.exists(video_path), f"Test video not found: {video_path}"
-            video_provider = VideoProvider(dev_name="test_video", video_source=video_path)
+            video_provider = VideoProvider(
+                dev_name="test_video", video_source=video_path, pool_scheduler=test_scheduler
+            )
 
             video_stream = video_provider.capture_video_as_observable(realtime=False, fps=15)
 
@@ -146,6 +150,8 @@ def on_completed() -> None:
 
         except Exception as e:
             pytest.fail(f"Test failed with error: {e}")
+        finally:
+            test_scheduler.executor.shutdown(wait=True)
 
     def test_clip_embedding_similarity(self) -> None:
         """Test CLIP embedding similarity search and text-to-image queries."""
@@ -205,7 +211,3 @@ def test_clip_embedding_similarity(self) -> None:
 
         except Exception as e:
             pytest.fail(f"Similarity test failed with error: {e}")
-
-
-if __name__ == "__main__":
-    pytest.main(["-v", "--disable-warnings", __file__])
diff --git a/dimos/conftest.py b/dimos/conftest.py
index 5d1ca2b860..2e8b558e6e 100644
--- a/dimos/conftest.py
+++ b/dimos/conftest.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import asyncio
+import os
 import threading
 
 from dotenv import load_dotenv
@@ -23,27 +24,28 @@
 load_dotenv()
 
 
-def _has_cuda():
-    try:
-        import torch
-    except Exception:
-        return False
-
-    try:
-        return bool(torch.cuda.is_available())
-    except Exception:
-        return False
+def pytest_configure(config):
+    config.addinivalue_line("markers", "tool: dev tooling")
+    config.addinivalue_line("markers", "slow: tests that are too slow for the fast loop")
+    config.addinivalue_line("markers", "mujoco: tests which open mujoco")
+    config.addinivalue_line("markers", "skipif_in_ci: skip when CI env var is set")
+    config.addinivalue_line("markers", "skipif_no_openai: skip when OPENAI_API_KEY is not set")
+    config.addinivalue_line("markers", "skipif_no_alibaba: skip when ALIBABA_API_KEY is not set")
 
 
 @pytest.hookimpl()
 def pytest_collection_modifyitems(config, items):
-    if not _has_cuda():
-        skip_marker = pytest.mark.skip(
-            reason="CUDA is not available (torch.cuda.is_available() returned False)"
-        )
-        for item in items:
-            if item.get_closest_marker("cuda"):
-                item.add_marker(skip_marker)
+    _skipif_markers = {
+        "skipif_in_ci": (bool(os.getenv("CI")), "Skipped in CI"),
+        "skipif_no_openai": (not os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY not set"),
+        "skipif_no_alibaba": (not os.getenv("ALIBABA_API_KEY"), "ALIBABA_API_KEY not set"),
+    }
+    for marker_name, (condition, reason) in _skipif_markers.items():
+        if condition:
+            skip = pytest.mark.skip(reason=reason)
+            for item in items:
+                if item.get_closest_marker(marker_name):
+                    item.add_marker(skip)
 
 
 @pytest.fixture
@@ -70,8 +72,6 @@ def _autoconf(request):
 _seen_threads_lock = threading.RLock()
 _before_test_threads = {}  # Map test name to set of thread IDs before test
 
-_skip_for = ["lcm", "heavy", "ros"]
-
 
 @pytest.fixture(scope="module")
 def dimos_cluster():
@@ -107,11 +107,6 @@ def pytest_sessionfinish(session):
 
 @pytest.fixture(autouse=True)
 def monitor_threads(request):
-    # Skip monitoring for tests marked with specified markers
-    if any(request.node.get_closest_marker(marker) for marker in _skip_for):
-        yield
-        return
-
     # Capture threads before test runs
     test_name = request.node.nodeid
     with _seen_threads_lock:
diff --git a/dimos/core/test_blueprints.py b/dimos/core/test_blueprints.py
index 09144054c1..fd18fe72d8 100644
--- a/dimos/core/test_blueprints.py
+++ b/dimos/core/test_blueprints.py
@@ -175,7 +175,7 @@ def test_global_config() -> None:
     assert blueprint_set.global_config_overrides["option2"] == 42
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_build_happy_path() -> None:
     pubsub.lcm.autoconf()
 
@@ -286,7 +286,7 @@ class Module3(Module):
     blueprint_set_remapped._verify_no_name_conflicts()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_remapping() -> None:
     """Test that remapping streams works correctly."""
     pubsub.lcm.autoconf()
@@ -355,7 +355,7 @@ def test_future_annotations_support() -> None:
     assert in_blueprint.streams[0] == StreamRef(name="data", type=FutureData, direction="in")
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_future_annotations_autoconnect() -> None:
     """Test that autoconnect works with modules using `from __future__ import annotations`."""
 
@@ -448,7 +448,7 @@ def start(self) -> None:
     def stop(self) -> None: ...
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_module_ref_direct() -> None:
     coordinator = autoconnect(
         Calculator1.blueprint(),
@@ -464,7 +464,7 @@ def test_module_ref_direct() -> None:
         coordinator.stop()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_module_ref_spec() -> None:
     coordinator = autoconnect(
         Calculator1.blueprint(),
@@ -480,7 +480,7 @@ def test_module_ref_spec() -> None:
         coordinator.stop()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_module_ref_remap_ambiguous() -> None:
     coordinator = (
         autoconnect(
diff --git a/dimos/core/test_core.py b/dimos/core/test_core.py
index c229659b84..3866d55bdb 100644
--- a/dimos/core/test_core.py
+++ b/dimos/core/test_core.py
@@ -101,7 +101,8 @@ def test_classmethods() -> None:
     nav._close_module()
 
 
-@pytest.mark.module
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_basic_deployment(dimos) -> None:
     robot = dimos.deploy(MockRobotClient)
 
@@ -136,4 +137,7 @@ def test_basic_deployment(dimos) -> None:
     assert nav.odom_msg_count >= 8
     assert nav.lidar_msg_count >= 8
 
-    dimos.shutdown()
+    nav.stop()
+    nav.stop_rpc_client()
+    robot.stop_rpc_client()
+    dimos.close_all()
diff --git a/dimos/core/test_native_module.py b/dimos/core/test_native_module.py
index 8af63b0bf4..a022be0685 100644
--- a/dimos/core/test_native_module.py
+++ b/dimos/core/test_native_module.py
@@ -131,7 +131,7 @@ def test_manual(dimos_cluster: DimosCluster, args_file: str) -> None:
     }
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 def test_autoconnect(args_file: str) -> None:
     """autoconnect passes correct topic args to the native subprocess."""
     blueprint = autoconnect(
diff --git a/dimos/core/test_stream.py b/dimos/core/test_stream.py
index 836f879b67..7a594f64e4 100644
--- a/dimos/core/test_stream.py
+++ b/dimos/core/test_stream.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 from collections.abc import Callable
+import threading
 import time
 
 import pytest
@@ -21,6 +22,7 @@
     In,
     LCMTransport,
     Module,
+    pLCMTransport,
     rpc,
 )
 from dimos.core.testing import MockRobotClient, dimos
@@ -37,14 +39,32 @@ class SubscriberBase(Module):
     def __init__(self) -> None:
         self.sub1_msgs = []
         self.sub2_msgs = []
+        self._sub1_received = threading.Event()
+        self._sub2_received = threading.Event()
         super().__init__()
 
+    def _sub1_callback(self, msg) -> None:
+        self.sub1_msgs.append(msg)
+        self._sub1_received.set()
+
+    def _sub2_callback(self, msg) -> None:
+        self.sub2_msgs.append(msg)
+        self._sub2_received.set()
+
     @rpc
     def sub1(self) -> None: ...
 
     @rpc
     def sub2(self) -> None: ...
 
+    @rpc
+    def wait_for_sub1_msg(self, timeout: float = 10) -> bool:
+        return self._sub1_received.wait(timeout)
+
+    @rpc
+    def wait_for_sub2_msg(self, timeout: float = 10) -> bool:
+        return self._sub2_received.wait(timeout)
+
     @rpc
     def active_subscribers(self):
         return self.odom.transport.active_subscribers
@@ -65,14 +85,14 @@ class ClassicSubscriber(SubscriberBase):
 
     @rpc
     def sub1(self) -> None:
-        self.unsub = self.odom.subscribe(self.sub1_msgs.append)
+        self.unsub = self.odom.subscribe(self._sub1_callback)
 
     @rpc
     def sub2(self) -> None:
-        self.unsub2 = self.odom.subscribe(self.sub2_msgs.append)
+        self.unsub2 = self.odom.subscribe(self._sub2_callback)
 
     @rpc
-    def stop(self) -> None:
+    def unsub_all(self) -> None:
         if self.unsub:
             self.unsub()
             self.unsub = None
@@ -90,14 +110,14 @@ class RXPYSubscriber(SubscriberBase):
 
     @rpc
     def sub1(self) -> None:
-        self.unsub = self.odom.observable().subscribe(self.sub1_msgs.append)
+        self.unsub = self.odom.observable().subscribe(self._sub1_callback)
 
     @rpc
     def sub2(self) -> None:
-        self.unsub2 = self.odom.observable().subscribe(self.sub2_msgs.append)
+        self.unsub2 = self.odom.observable().subscribe(self._sub2_callback)
 
     @rpc
-    def stop(self) -> None:
+    def unsub_all(self) -> None:
         if self.unsub:
             self.unsub.dispose()
             self.unsub = None
@@ -153,12 +173,13 @@ def wrapped_unsubscribe() -> None:
 
 
 @pytest.mark.parametrize("subscriber_class", [ClassicSubscriber, RXPYSubscriber])
-@pytest.mark.module
+@pytest.mark.slow
 def test_subscription(dimos, subscriber_class) -> None:
     robot = dimos.deploy(MockRobotClient)
 
     robot.lidar.transport = SpyLCMTransport("/lidar", PointCloud2)
     robot.odometry.transport = SpyLCMTransport("/odom", Odometry)
+    robot.mov.transport = pLCMTransport("/mov")
 
     subscriber = dimos.deploy(subscriber_class)
 
@@ -166,16 +187,16 @@ def test_subscription(dimos, subscriber_class) -> None:
 
     robot.start()
     subscriber.sub1()
-    time.sleep(0.25)
+    subscriber.wait_for_sub1_msg()
 
     assert subscriber.sub1_msgs_len() > 0
     assert subscriber.sub2_msgs_len() == 0
     assert subscriber.active_subscribers() == 1
 
     subscriber.sub2()
+    subscriber.wait_for_sub2_msg()
 
-    time.sleep(0.25)
-    subscriber.stop()
+    subscriber.unsub_all()
 
     assert subscriber.active_subscribers() == 0
     assert subscriber.sub1_msgs_len() != 0
@@ -183,20 +204,24 @@ def test_subscription(dimos, subscriber_class) -> None:
 
     total_msg_n = subscriber.sub1_msgs_len() + subscriber.sub2_msgs_len()
 
-    time.sleep(0.25)
+    time.sleep(0.5)
 
     # ensuring no new messages have passed through
     assert total_msg_n == subscriber.sub1_msgs_len() + subscriber.sub2_msgs_len()
 
     robot.stop()
+    subscriber.stop_rpc_client()
+    robot.stop_rpc_client()
+    dimos.close_all()
 
 
-@pytest.mark.module
+@pytest.mark.slow
 def test_get_next(dimos) -> None:
     robot = dimos.deploy(MockRobotClient)
 
     robot.lidar.transport = SpyLCMTransport("/lidar", PointCloud2)
     robot.odometry.transport = SpyLCMTransport("/odom", Odometry)
+    robot.mov.transport = pLCMTransport("/mov")
 
     subscriber = dimos.deploy(RXPYSubscriber)
     subscriber.odom.connect(robot.odometry)
@@ -218,14 +243,18 @@ def test_get_next(dimos) -> None:
 
     assert next_odom != odom
     robot.stop()
+    subscriber.stop_rpc_client()
+    robot.stop_rpc_client()
+    dimos.close_all()
 
 
-@pytest.mark.module
+@pytest.mark.slow
 def test_hot_getter(dimos) -> None:
     robot = dimos.deploy(MockRobotClient)
 
     robot.lidar.transport = SpyLCMTransport("/lidar", PointCloud2)
     robot.odometry.transport = SpyLCMTransport("/odom", Odometry)
+    robot.mov.transport = pLCMTransport("/mov")
 
     subscriber = dimos.deploy(RXPYSubscriber)
     subscriber.odom.connect(robot.odometry)
@@ -254,3 +283,6 @@ def test_hot_getter(dimos) -> None:
     subscriber.stop_hot_getter()
 
     robot.stop()
+    subscriber.stop_rpc_client()
+    robot.stop_rpc_client()
+    dimos.close_all()
diff --git a/dimos/core/test_worker.py b/dimos/core/test_worker.py
index 98a7c5782d..6892d226fd 100644
--- a/dimos/core/test_worker.py
+++ b/dimos/core/test_worker.py
@@ -82,7 +82,7 @@ def worker_manager():
         manager.close_all()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_worker_manager_basic(worker_manager):
     module = worker_manager.deploy(SimpleModule)
     module.start()
@@ -99,7 +99,7 @@ def test_worker_manager_basic(worker_manager):
     module.stop()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_worker_manager_multiple_different_modules(worker_manager):
     module1 = worker_manager.deploy(SimpleModule)
     module2 = worker_manager.deploy(AnotherModule)
@@ -120,7 +120,7 @@ def test_worker_manager_multiple_different_modules(worker_manager):
     module2.stop()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_worker_manager_parallel_deployment(worker_manager):
     modules = worker_manager.deploy_parallel(
         [
diff --git a/dimos/e2e_tests/test_control_coordinator.py b/dimos/e2e_tests/test_control_coordinator.py
index f6e520831d..5bb7a096f7 100644
--- a/dimos/e2e_tests/test_control_coordinator.py
+++ b/dimos/e2e_tests/test_control_coordinator.py
@@ -18,7 +18,6 @@
 Unlike unit tests, these verify the full system integration.
 """
 
-import os
 import time
 
 import pytest
@@ -29,8 +28,8 @@
 from dimos.msgs.trajectory_msgs import JointTrajectory, TrajectoryPoint, TrajectoryState
 
 
-@pytest.mark.skipif(bool(os.getenv("CI")), reason="LCM doesn't work in CI.")
-@pytest.mark.e2e
+@pytest.mark.skipif_in_ci
+@pytest.mark.slow
 class TestControlCoordinatorE2E:
     """End-to-end tests for ControlCoordinator."""
 
diff --git a/dimos/e2e_tests/test_dimos_cli_e2e.py b/dimos/e2e_tests/test_dimos_cli_e2e.py
index ede0ec7a3a..f27502d620 100644
--- a/dimos/e2e_tests/test_dimos_cli_e2e.py
+++ b/dimos/e2e_tests/test_dimos_cli_e2e.py
@@ -12,14 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
-
 import pytest
 
 
-@pytest.mark.skipif(bool(os.getenv("CI")), reason="LCM spy doesn't work in CI.")
-@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set.")
-@pytest.mark.e2e
+@pytest.mark.skipif_in_ci
+@pytest.mark.skipif_no_openai
+@pytest.mark.slow
 def test_dimos_skills(lcm_spy, start_blueprint, human_input) -> None:
     lcm_spy.save_topic("/agent")
     lcm_spy.save_topic("/rpc/Agent/on_system_modules/res")
diff --git a/dimos/e2e_tests/test_person_follow.py b/dimos/e2e_tests/test_person_follow.py
index abb9cfb4fa..090ee90f2a 100644
--- a/dimos/e2e_tests/test_person_follow.py
+++ b/dimos/e2e_tests/test_person_follow.py
@@ -13,7 +13,6 @@
 # limitations under the License.
 
 from collections.abc import Callable, Generator
-import os
 import threading
 import time
 
@@ -53,8 +52,8 @@ def run_person_track() -> None:
         publisher.stop()
 
 
-@pytest.mark.skipif(bool(os.getenv("CI")), reason="LCM spy doesn't work in CI.")
-@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set.")
+@pytest.mark.skipif_in_ci
+@pytest.mark.skipif_no_openai
 @pytest.mark.mujoco
 def test_person_follow(
     lcm_spy: LcmSpy,
diff --git a/dimos/e2e_tests/test_simulation_module.py b/dimos/e2e_tests/test_simulation_module.py
index 6c15f62056..b5902ad7e2 100644
--- a/dimos/e2e_tests/test_simulation_module.py
+++ b/dimos/e2e_tests/test_simulation_module.py
@@ -14,8 +14,6 @@
 
 """End-to-end tests for the simulation module."""
 
-import os
-
 import pytest
 
 from dimos.msgs.sensor_msgs import JointCommand, JointState, RobotState
@@ -31,8 +29,8 @@ def _positions_within_tolerance(
     return all(abs(positions[i] - target[i]) <= tolerance for i in range(len(target)))
 
 
-@pytest.mark.skipif(bool(os.getenv("CI")), reason="LCM doesn't work in CI.")
-@pytest.mark.e2e
+@pytest.mark.skipif_in_ci
+@pytest.mark.slow
 class TestSimulationModuleE2E:
     def test_xarm7_joint_state_published(self, lcm_spy, start_blueprint) -> None:
         joint_state_topic = "/xarm/joint_states#sensor_msgs.JointState"
diff --git a/dimos/e2e_tests/test_spatial_memory.py b/dimos/e2e_tests/test_spatial_memory.py
index 8b03a9915c..ad22368678 100644
--- a/dimos/e2e_tests/test_spatial_memory.py
+++ b/dimos/e2e_tests/test_spatial_memory.py
@@ -14,7 +14,6 @@
 
 from collections.abc import Callable
 import math
-import os
 import time
 
 import pytest
@@ -23,8 +22,8 @@
 from dimos.e2e_tests.lcm_spy import LcmSpy
 
 
-@pytest.mark.skipif(bool(os.getenv("CI")), reason="LCM spy doesn't work in CI.")
-@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set.")
+@pytest.mark.skipif_in_ci
+@pytest.mark.skipif_no_openai
 @pytest.mark.mujoco
 def test_spatial_memory_navigation(
     lcm_spy: LcmSpy,
diff --git a/dimos/manipulation/planning/monitor/world_obstacle_monitor.py b/dimos/manipulation/planning/monitor/world_obstacle_monitor.py
index 6082ab93a9..a96d3efaf6 100644
--- a/dimos/manipulation/planning/monitor/world_obstacle_monitor.py
+++ b/dimos/manipulation/planning/monitor/world_obstacle_monitor.py
@@ -555,7 +555,7 @@ def list_added_obstacles(self) -> list[dict[str, Any]]:
                 entry = self._object_cache.get(oid)
                 if entry is None:
                     continue
-                obj, first_seen, last_seen = entry
+                obj, _first_seen, _last_seen = entry
                 if not isinstance(obj, Object):
                     continue
                 result.append(
diff --git a/dimos/mapping/occupancy/test_extrude_occupancy.py b/dimos/mapping/occupancy/test_extrude_occupancy.py
index 88f05d7780..76d714a2a1 100644
--- a/dimos/mapping/occupancy/test_extrude_occupancy.py
+++ b/dimos/mapping/occupancy/test_extrude_occupancy.py
@@ -18,7 +18,7 @@
 from dimos.utils.data import get_data
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_generate_mujoco_scene(occupancy) -> None:
     with open(get_data("expected_occupancy_scene.xml")) as f:
         expected = f.read()
diff --git a/dimos/memory/timeseries/test_legacy.py b/dimos/memory/timeseries/test_legacy.py
index aaad962a95..c77ec64a76 100644
--- a/dimos/memory/timeseries/test_legacy.py
+++ b/dimos/memory/timeseries/test_legacy.py
@@ -13,12 +13,16 @@
 # limitations under the License.
 """Tests specific to LegacyPickleStore."""
 
+import pytest
+
 from dimos.memory.timeseries.legacy import LegacyPickleStore
 
 
 class TestLegacyPickleStoreRealData:
     """Test LegacyPickleStore with real recorded data."""
 
+    @pytest.mark.skipif_in_ci
+    @pytest.mark.slow
     def test_read_lidar_recording(self) -> None:
         """Test reading from unitree_go2_bigoffice/lidar recording."""
         store = LegacyPickleStore("unitree_go2_bigoffice/lidar")
diff --git a/dimos/models/embedding/test_embedding.py b/dimos/models/embedding/test_embedding.py
index a87a2f5a57..466c974b32 100644
--- a/dimos/models/embedding/test_embedding.py
+++ b/dimos/models/embedding/test_embedding.py
@@ -20,7 +20,8 @@
     ],
     ids=["clip", "mobileclip", "treid"],
 )
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_embedding_model(model_class: type, model_name: str, supports_text: bool) -> None:
     """Test embedding functionality across different model types."""
     image = Image.from_file(get_data("cafe.jpg")).to_rgb()
@@ -94,7 +95,8 @@ def test_embedding_model(model_class: type, model_name: str, supports_text: bool
     ],
     ids=["clip", "mobileclip"],
 )
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_text_image_retrieval(model_class: type, model_name: str) -> None:
     """Test text-to-image retrieval using embedding similarity."""
     image = Image.from_file(get_data("cafe.jpg")).to_rgb()
@@ -126,7 +128,8 @@ def test_text_image_retrieval(model_class: type, model_name: str) -> None:
     print(f"\n{model_name} retrieval test passed!")
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_embedding_device_transfer() -> None:
     """Test embedding device transfer operations."""
     image = Image.from_file(get_data("cafe.jpg")).to_rgb()
diff --git a/dimos/models/vl/test_base.py b/dimos/models/vl/test_base.py
index a7296bd87b..2e4229d944 100644
--- a/dimos/models/vl/test_base.py
+++ b/dimos/models/vl/test_base.py
@@ -1,4 +1,3 @@
-import os
 from unittest.mock import MagicMock
 
 from dimos_lcm.foxglove_msgs.ImageAnnotations import ImageAnnotations
@@ -78,7 +77,7 @@ def test_query_detections_mocked() -> None:
 
 
 @pytest.mark.tool
-@pytest.mark.skipif(not os.getenv("ALIBABA_API_KEY"), reason="ALIBABA_API_KEY not set")
+@pytest.mark.skipif_no_alibaba
 def test_query_detections_real() -> None:
     """Test query_detections with real API calls (requires API key)."""
     # Load test image
diff --git a/dimos/models/vl/test_captioner.py b/dimos/models/vl/test_captioner.py
index 081f3bcefc..c7ebb8fc63 100644
--- a/dimos/models/vl/test_captioner.py
+++ b/dimos/models/vl/test_captioner.py
@@ -44,7 +44,8 @@ def florence2_model(request: pytest.FixtureRequest) -> Generator[Florence2Model,
     yield from generic_model_fixture(request.param)
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_captioner(captioner_model: CaptionerModel, test_image: Image) -> None:
     """Test captioning functionality across different model types."""
     # Test single caption
@@ -72,7 +73,8 @@ def test_captioner(captioner_model: CaptionerModel, test_image: Image) -> None:
     assert all(isinstance(c, str) and len(c) > 0 for c in captions)
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_florence2_detail_levels(florence2_model: Florence2Model, test_image: Image) -> None:
     """Test Florence-2 different detail levels."""
     detail_levels = ["brief", "normal", "detailed", "more_detailed"]
diff --git a/dimos/models/vl/test_vlm.py b/dimos/models/vl/test_vlm.py
index 741e0dede2..54ceddadc5 100644
--- a/dimos/models/vl/test_vlm.py
+++ b/dimos/models/vl/test_vlm.py
@@ -32,7 +32,8 @@
         (QwenVlModel, "Qwen"),
     ],
 )
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_vlm_bbox_detections(model_class: "type[VlModel]", model_name: str) -> None:
     if model_class is MoondreamHostedVlModel and 'MOONDREAM_API_KEY' not in os.environ:
         pytest.skip("Need MOONDREAM_API_KEY to run")
@@ -104,7 +105,8 @@ def test_vlm_bbox_detections(model_class: "type[VlModel]", model_name: str) -> N
         (QwenVlModel, "Qwen"),
     ],
 )
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_vlm_point_detections(model_class: "type[VlModel]", model_name: str) -> None:
     """Test VLM point detection capabilities."""
 
@@ -172,7 +174,8 @@ def test_vlm_point_detections(model_class: "type[VlModel]", model_name: str) ->
         (MoondreamVlModel, "Moondream"),
     ],
 )
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_vlm_query_multi(model_class: "type[VlModel]", model_name: str) -> None:
     """Test query_multi optimization - single image, multiple queries."""
     image = Image.from_file(get_data("cafe.jpg")).to_rgb()
@@ -222,7 +225,7 @@ def test_vlm_query_multi(model_class: "type[VlModel]", model_name: str) -> None:
     ],
 )
 @pytest.mark.tool
-@pytest.mark.gpu
+@pytest.mark.slow
 def test_vlm_query_batch(model_class: "type[VlModel]", model_name: str) -> None:
     """Test query_batch optimization - multiple images, same query."""
     from dimos.utils.testing import TimedSensorReplay
@@ -275,7 +278,8 @@ def test_vlm_query_batch(model_class: "type[VlModel]", model_name: str) -> None:
         (QwenVlModel, [None, (512, 512), (256, 256)]),
     ],
 )
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_vlm_resize(
     model_class: "type[VlModel]",
     sizes: list[tuple[int, int] | None],
diff --git a/dimos/msgs/geometry_msgs/test_TwistStamped.py b/dimos/msgs/geometry_msgs/test_TwistStamped.py
index afb8489032..d37b4b2717 100644
--- a/dimos/msgs/geometry_msgs/test_TwistStamped.py
+++ b/dimos/msgs/geometry_msgs/test_TwistStamped.py
@@ -52,15 +52,3 @@ def test_pickle_encode_decode() -> None:
     assert isinstance(twist_dest, TwistStamped)
     assert twist_dest is not twist_source
     assert twist_dest == twist_source
-
-
-if __name__ == "__main__":
-    print("Running test_lcm_encode_decode...")
-    test_lcm_encode_decode()
-    print("test_lcm_encode_decode passed")
-
-    print("Running test_pickle_encode_decode...")
-    test_pickle_encode_decode()
-    print("test_pickle_encode_decode passed")
-
-    print("\nAll tests passed!")
diff --git a/dimos/msgs/nav_msgs/test_OccupancyGrid.py b/dimos/msgs/nav_msgs/test_OccupancyGrid.py
index 29ef196de8..d1ec8938b4 100644
--- a/dimos/msgs/nav_msgs/test_OccupancyGrid.py
+++ b/dimos/msgs/nav_msgs/test_OccupancyGrid.py
@@ -26,7 +26,6 @@
 from dimos.msgs.geometry_msgs import Pose
 from dimos.msgs.nav_msgs import OccupancyGrid
 from dimos.msgs.sensor_msgs import PointCloud2
-from dimos.protocol.pubsub.impl.lcmpubsub import LCM, Topic
 from dimos.utils.data import get_data
 
 
@@ -364,107 +363,3 @@ def test_max() -> None:
     assert maxed.unknown_cells == 3  # Same as original
     assert maxed.occupied_cells == 13  # All non-unknown cells
     assert maxed.free_cells == 0  # No free cells
-
-
-@pytest.mark.lcm
-def test_lcm_broadcast() -> None:
-    """Test broadcasting OccupancyGrid and gradient over LCM."""
-    file_path = get_data("lcm_msgs") / "sensor_msgs/PointCloud2.pickle"
-    with open(file_path, "rb") as f:
-        lcm_msg = pickle.loads(f.read())
-
-    pointcloud = PointCloud2.lcm_decode(lcm_msg)
-
-    # Create occupancy grid from pointcloud
-    occupancygrid = general_occupancy(pointcloud, resolution=0.05, min_height=0.1, max_height=2.0)
-    # Apply inflation separately if needed
-    occupancygrid = simple_inflate(occupancygrid, 0.1)
-
-    # Create gradient field with larger max_distance for better visualization
-    gradient_grid = gradient(occupancygrid, obstacle_threshold=70, max_distance=2.0)
-
-    # Debug: Print actual values to see the difference
-    print("\n=== DEBUG: Comparing grids ===")
-    print(f"Original grid unique values: {np.unique(occupancygrid.grid)}")
-    print(f"Gradient grid unique values: {np.unique(gradient_grid.grid)}")
-
-    # Find an area with occupied cells to show the difference
-    occupied_indices = np.argwhere(occupancygrid.grid == 100)
-    if len(occupied_indices) > 0:
-        # Pick a point near an occupied cell
-        idx = len(occupied_indices) // 2  # Middle occupied cell
-        sample_y, sample_x = occupied_indices[idx]
-        sample_size = 15
-
-        # Ensure we don't go out of bounds
-        y_start = max(0, sample_y - sample_size // 2)
-        y_end = min(occupancygrid.height, y_start + sample_size)
-        x_start = max(0, sample_x - sample_size // 2)
-        x_end = min(occupancygrid.width, x_start + sample_size)
-
-        print(f"\nSample area around occupied cell ({sample_x}, {sample_y}):")
-        print("Original occupancy grid:")
-        print(occupancygrid.grid[y_start:y_end, x_start:x_end])
-        print("\nGradient grid (same area):")
-        print(gradient_grid.grid[y_start:y_end, x_start:x_end])
-    else:
-        print("\nNo occupied cells found for sampling")
-
-    # Check statistics
-    print("\nOriginal grid stats:")
-    print(f"  Occupied (100): {np.sum(occupancygrid.grid == 100)} cells")
-    print(f"  Inflated (99): {np.sum(occupancygrid.grid == 99)} cells")
-    print(f"  Free (0): {np.sum(occupancygrid.grid == 0)} cells")
-    print(f"  Unknown (-1): {np.sum(occupancygrid.grid == -1)} cells")
-
-    print("\nGradient grid stats:")
-    print(f"  Max gradient (100): {np.sum(gradient_grid.grid == 100)} cells")
-    print(
-        f"  High gradient (80-99): {np.sum((gradient_grid.grid >= 80) & (gradient_grid.grid < 100))} cells"
-    )
-    print(
-        f"  Medium gradient (40-79): {np.sum((gradient_grid.grid >= 40) & (gradient_grid.grid < 80))} cells"
-    )
-    print(
-        f"  Low gradient (1-39): {np.sum((gradient_grid.grid >= 1) & (gradient_grid.grid < 40))} cells"
-    )
-    print(f"  Zero gradient (0): {np.sum(gradient_grid.grid == 0)} cells")
-    print(f"  Unknown (-1): {np.sum(gradient_grid.grid == -1)} cells")
-
-    # # Save debug images
-    # import matplotlib.pyplot as plt
-
-    # fig, axes = plt.subplots(1, 2, figsize=(12, 5))
-
-    # # Original
-    # ax = axes[0]
-    # im1 = ax.imshow(occupancygrid.grid, origin="lower", cmap="gray_r", vmin=-1, vmax=100)
-    # ax.set_title(f"Original Occupancy Grid\n{occupancygrid}")
-    # plt.colorbar(im1, ax=ax)
-
-    # # Gradient
-    # ax = axes[1]
-    # im2 = ax.imshow(gradient_grid.grid, origin="lower", cmap="hot", vmin=-1, vmax=100)
-    # ax.set_title(f"Gradient Grid\n{gradient_grid}")
-    # plt.colorbar(im2, ax=ax)
-
-    # plt.tight_layout()
-    # plt.savefig("lcm_debug_grids.png", dpi=150)
-    # print("\nSaved debug visualization to lcm_debug_grids.png")
-    # plt.close()
-
-    # Broadcast all the data
-    lcm = LCM()
-    lcm.start()
-    lcm.publish(Topic("/global_map", PointCloud2), pointcloud)
-    lcm.publish(Topic("/global_costmap", OccupancyGrid), occupancygrid)
-    lcm.publish(Topic("/global_gradient", OccupancyGrid), gradient_grid)
-
-    print("\nPublished to LCM:")
-    print(f"  /global_map: PointCloud2 with {len(pointcloud)} points")
-    print(f"  /global_costmap: {occupancygrid}")
-    print(f"  /global_gradient: {gradient_grid}")
-    print("\nGradient info:")
-    print("  Values: 0 (free far from obstacles) -> 100 (at obstacles)")
-    print(f"  Unknown cells: {gradient_grid.unknown_cells} (preserved as -1)")
-    print("  Max distance for gradient: 5.0 meters")
diff --git a/dimos/msgs/sensor_msgs/test_PointCloud2.py b/dimos/msgs/sensor_msgs/test_PointCloud2.py
index 501a4cd441..f48802ab7a 100644
--- a/dimos/msgs/sensor_msgs/test_PointCloud2.py
+++ b/dimos/msgs/sensor_msgs/test_PointCloud2.py
@@ -153,8 +153,3 @@ def test_bounding_box_intersects() -> None:
         pass
 
     print("✓ All bounding box intersection tests passed!")
-
-
-if __name__ == "__main__":
-    test_lcm_encode_decode()
-    test_bounding_box_intersects()
diff --git a/dimos/perception/detection/reid/test_embedding_id_system.py b/dimos/perception/detection/reid/test_embedding_id_system.py
index b9e6f591ee..cc8632627f 100644
--- a/dimos/perception/detection/reid/test_embedding_id_system.py
+++ b/dimos/perception/detection/reid/test_embedding_id_system.py
@@ -46,7 +46,8 @@ def test_image():
     return Image.from_file(get_data("cafe.jpg")).to_rgb()
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_update_embedding_single(track_associator, mobileclip_model, test_image) -> None:
     """Test updating embedding for a single track."""
     embedding = mobileclip_model.embed(test_image)
@@ -64,7 +65,8 @@ def test_update_embedding_single(track_associator, mobileclip_model, test_image)
     assert abs(norm - 1.0) < 0.01, "Embedding should be normalized"
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_update_embedding_multiple(track_associator, mobileclip_model, test_image) -> None:
     """Test storing multiple embeddings per track."""
     embedding1 = mobileclip_model.embed(test_image)
@@ -91,7 +93,8 @@ def test_update_embedding_multiple(track_associator, mobileclip_model, test_imag
     assert similarity > 0.99, "Same image should produce very similar embeddings"
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_negative_constraints(track_associator) -> None:
     """Test negative constraint recording."""
     # Simulate frame with 3 tracks
@@ -107,7 +110,8 @@ def test_negative_constraints(track_associator) -> None:
     assert 2 in track_associator.negative_pairs[3]
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_associate_new_track(track_associator, mobileclip_model, test_image) -> None:
     """Test associating a new track creates new long_term_id."""
     embedding = mobileclip_model.embed(test_image)
@@ -121,7 +125,8 @@ def test_associate_new_track(track_associator, mobileclip_model, test_image) ->
     assert track_associator.long_term_counter == 1
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_associate_similar_tracks(track_associator, mobileclip_model, test_image) -> None:
     """Test associating similar tracks to same long_term_id."""
     # Create embeddings from same image (should be very similar)
@@ -141,7 +146,8 @@ def test_associate_similar_tracks(track_associator, mobileclip_model, test_image
     assert track_associator.long_term_counter == 1, "Only one long_term_id should be created"
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_associate_with_negative_constraint(track_associator, mobileclip_model, test_image) -> None:
     """Test that negative constraints prevent association."""
     # Create similar embeddings
@@ -166,7 +172,8 @@ def test_associate_with_negative_constraint(track_associator, mobileclip_model,
     assert track_associator.long_term_counter == 2, "Two long_term_ids should be created"
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_associate_different_objects(track_associator, mobileclip_model, test_image) -> None:
     """Test that dissimilar embeddings get different long_term_ids."""
     # Create embeddings for image and text (very different)
@@ -186,7 +193,8 @@ def test_associate_different_objects(track_associator, mobileclip_model, test_im
     assert track_associator.long_term_counter == 2
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_associate_returns_cached(track_associator, mobileclip_model, test_image) -> None:
     """Test that repeated calls return same long_term_id."""
     embedding = mobileclip_model.embed(test_image)
@@ -202,7 +210,8 @@ def test_associate_returns_cached(track_associator, mobileclip_model, test_image
     assert track_associator.long_term_counter == 1, "Should not create new ID"
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_associate_no_embedding(track_associator) -> None:
     """Test that associate creates new ID for track without embedding."""
     # Track with no embedding gets assigned a new ID
@@ -211,7 +220,8 @@ def test_associate_no_embedding(track_associator) -> None:
     assert track_associator.long_term_counter == 1
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_embeddings_stored_as_numpy(track_associator, mobileclip_model, test_image) -> None:
     """Test that embeddings are stored as numpy arrays for efficient CPU comparisons."""
     embedding = mobileclip_model.embed(test_image)
@@ -232,7 +242,8 @@ def test_embeddings_stored_as_numpy(track_associator, mobileclip_model, test_ima
         assert isinstance(emb, np.ndarray)
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_similarity_threshold_configurable(mobileclip_model) -> None:
     """Test that similarity threshold is configurable."""
     associator_strict = EmbeddingIDSystem(model=lambda: mobileclip_model, similarity_threshold=0.95)
@@ -242,7 +253,8 @@ def test_similarity_threshold_configurable(mobileclip_model) -> None:
     assert associator_loose.similarity_threshold == 0.50
 
 
-@pytest.mark.gpu
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 def test_multi_track_scenario(track_associator, mobileclip_model, test_image) -> None:
     """Test realistic scenario with multiple tracks across frames."""
     # Frame 1: Track 1 appears
diff --git a/dimos/perception/detection/test_moduleDB.py b/dimos/perception/detection/test_moduleDB.py
deleted file mode 100644
index 23885a1c60..0000000000
--- a/dimos/perception/detection/test_moduleDB.py
+++ /dev/null
@@ -1,59 +0,0 @@
-# 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.
-import time
-
-from lcm_msgs.foxglove_msgs import SceneUpdate
-import pytest
-
-from dimos.core import LCMTransport
-from dimos.msgs.foxglove_msgs import ImageAnnotations
-from dimos.msgs.geometry_msgs import PoseStamped
-from dimos.msgs.sensor_msgs import Image, PointCloud2
-from dimos.msgs.vision_msgs import Detection2DArray
-from dimos.perception.detection.moduleDB import ObjectDBModule
-from dimos.robot.unitree.go2 import connection as go2_connection
-
-
-@pytest.mark.module
-def test_moduleDB(dimos_cluster) -> None:
-    connection = go2_connection.deploy(dimos_cluster, "fake")
-
-    moduleDB = dimos_cluster.deploy(
-        ObjectDBModule,
-        camera_info=go2_connection._camera_info_static(),
-        goto=lambda obj_id: print(f"Going to {obj_id}"),
-    )
-    moduleDB.image.connect(connection.color_image)
-    moduleDB.pointcloud.connect(connection.lidar)
-
-    moduleDB.annotations.transport = LCMTransport("/annotations", ImageAnnotations)
-    moduleDB.detections.transport = LCMTransport("/detections", Detection2DArray)
-
-    moduleDB.detected_pointcloud_0.transport = LCMTransport("/detected/pointcloud/0", PointCloud2)
-    moduleDB.detected_pointcloud_1.transport = LCMTransport("/detected/pointcloud/1", PointCloud2)
-    moduleDB.detected_pointcloud_2.transport = LCMTransport("/detected/pointcloud/2", PointCloud2)
-
-    moduleDB.detected_image_0.transport = LCMTransport("/detected/image/0", Image)
-    moduleDB.detected_image_1.transport = LCMTransport("/detected/image/1", Image)
-    moduleDB.detected_image_2.transport = LCMTransport("/detected/image/2", Image)
-
-    moduleDB.scene_update.transport = LCMTransport("/scene_update", SceneUpdate)
-    moduleDB.target.transport = LCMTransport("/target", PoseStamped)
-
-    connection.start()
-    moduleDB.start()
-
-    time.sleep(4)
-    print("VLM RES", moduleDB.navigate_to_object_in_view("white floor"))
-    time.sleep(30)
diff --git a/dimos/perception/experimental/temporal_memory/test_temporal_memory_module.py b/dimos/perception/experimental/temporal_memory/test_temporal_memory_module.py
index 1d0dab007b..ef584a2527 100644
--- a/dimos/perception/experimental/temporal_memory/test_temporal_memory_module.py
+++ b/dimos/perception/experimental/temporal_memory/test_temporal_memory_module.py
@@ -76,9 +76,9 @@ def stop(self) -> None:
         logger.info("VideoReplayModule stopped")
 
 
-@pytest.mark.skipif(bool(os.getenv("CI")), reason="LCM replay + dataset not CI-safe.")
-@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set.")
-@pytest.mark.neverending
+@pytest.mark.skipif_in_ci
+@pytest.mark.skipif_no_openai
+@pytest.mark.slow
 class TestTemporalMemoryModule:
     @pytest.fixture(scope="function")
     def temp_dir(self):
@@ -221,7 +221,3 @@ async def test_temporal_memory_module_with_replay(
         assert (output_path / "frames_index.jsonl").exists(), "frames_index.jsonl should exist"
 
         logger.info("All temporal memory module tests passed!")
-
-
-if __name__ == "__main__":
-    pytest.main(["-v", "-s", __file__])
diff --git a/dimos/perception/test_spatial_memory.py b/dimos/perception/test_spatial_memory.py
index d4b188ced3..433896aefe 100644
--- a/dimos/perception/test_spatial_memory.py
+++ b/dimos/perception/test_spatial_memory.py
@@ -20,13 +20,14 @@
 import numpy as np
 import pytest
 from reactivex import operators as ops
+from reactivex.scheduler import ThreadPoolScheduler
 
 from dimos.msgs.geometry_msgs import Pose
 from dimos.perception.spatial_perception import SpatialMemory
 from dimos.stream.video_provider import VideoProvider
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 class TestSpatialMemory:
     @pytest.fixture(scope="class")
     def temp_dir(self):
@@ -87,6 +88,7 @@ def test_image_embedding(self, spatial_memory) -> None:
 
     def test_spatial_memory_processing(self, spatial_memory, temp_dir) -> None:
         """Test processing video frames and building spatial memory with CLIP embeddings."""
+        test_scheduler = ThreadPoolScheduler(max_workers=4)
         try:
             # Use the shared spatial_memory fixture
             memory = spatial_memory
@@ -95,7 +97,9 @@ def test_spatial_memory_processing(self, spatial_memory, temp_dir) -> None:
 
             video_path = get_data("assets") / "trimmed_video_office.mov"
             assert os.path.exists(video_path), f"Test video not found: {video_path}"
-            video_provider = VideoProvider(dev_name="test_video", video_source=video_path)
+            video_provider = VideoProvider(
+                dev_name="test_video", video_source=video_path, pool_scheduler=test_scheduler
+            )
             video_stream = video_provider.capture_video_as_observable(realtime=False, fps=15)
 
             # Create a frame counter for position generation
@@ -196,7 +200,4 @@ def on_completed() -> None:
             pytest.fail(f"Error in test: {e}")
         finally:
             video_provider.dispose_all()
-
-
-if __name__ == "__main__":
-    pytest.main(["-v", __file__])
+            test_scheduler.executor.shutdown(wait=True)
diff --git a/dimos/perception/test_spatial_memory_module.py b/dimos/perception/test_spatial_memory_module.py
index 98ec7a1212..a8c42d4f0e 100644
--- a/dimos/perception/test_spatial_memory_module.py
+++ b/dimos/perception/test_spatial_memory_module.py
@@ -22,6 +22,7 @@
 
 from dimos import core
 from dimos.core import Module, Out, rpc
+from dimos.msgs.geometry_msgs import Transform
 from dimos.msgs.sensor_msgs import Image
 from dimos.perception.spatial_perception import SpatialMemory
 from dimos.robot.unitree.type.odometry import Odometry
@@ -70,29 +71,31 @@ def stop(self) -> None:
 
 
 class OdometryReplayModule(Module):
-    """Module that replays odometry data from TimedSensorReplay."""
-
-    odom_out: Out[Odometry]
+    """Module that replays odometry data and publishes to the tf system."""
 
     def __init__(self, odom_path: str) -> None:
         super().__init__()
         self.odom_path = odom_path
         self._subscription = None
 
+    def _publish_tf(self, odom: Odometry) -> None:
+        """Convert odometry to TF transforms and publish."""
+        self.tf.publish(Transform.from_pose("base_link", odom))
+
     @rpc
     def start(self) -> None:
         """Start replaying odometry data."""
         # Use TimedSensorReplay to replay odometry
         odom_replay = TimedSensorReplay(self.odom_path, autocast=Odometry.from_msg)
 
-        # Subscribe to the replay stream and publish to LCM
+        # Subscribe to the replay stream and publish to tf
         self._subscription = (
             odom_replay.stream()
             .pipe(
                 ops.sample(0.5),  # Sample every 500ms
                 ops.take(10),  # Only take 10 odometry updates total
             )
-            .subscribe(self.odom_out.publish)
+            .subscribe(self._publish_tf)
         )
 
         logger.info("OdometryReplayModule started")
@@ -106,8 +109,8 @@ def stop(self) -> None:
         logger.info("OdometryReplayModule stopped")
 
 
-@pytest.mark.gpu
-@pytest.mark.neverending
+@pytest.mark.slow
+@pytest.mark.skipif_in_ci
 class TestSpatialMemoryModule:
     @pytest.fixture(scope="function")
     def temp_dir(self):
@@ -135,9 +138,8 @@ async def test_spatial_memory_module_with_replay(self, temp_dir):
             video_module = dimos.deploy(VideoReplayModule, video_path)
             video_module.video_out.transport = core.LCMTransport("/test_video", Image)
 
-            # Odometry replay module
+            # Odometry replay module (publishes to tf system directly)
             odom_module = dimos.deploy(OdometryReplayModule, odom_path)
-            odom_module.odom_out.transport = core.LCMTransport("/test_odom", Odometry)
 
             # Spatial memory module
             spatial_memory = dimos.deploy(
@@ -153,9 +155,8 @@ async def test_spatial_memory_module_with_replay(self, temp_dir):
                 output_dir=os.path.join(temp_dir, "images"),
             )
 
-            # Connect streams
-            spatial_memory.video.connect(video_module.video_out)
-            spatial_memory.odom.connect(odom_module.odom_out)
+            # Connect video stream
+            spatial_memory.color_image.connect(video_module.video_out)
 
             # Start all modules
             video_module.start()
@@ -209,19 +210,12 @@ async def test_spatial_memory_module_with_replay(self, temp_dir):
 
             video_module.stop()
             odom_module.stop()
-            logger.info("Stopped replay modules")
+            spatial_memory.stop()
+            logger.info("Stopped all modules")
 
             logger.info("All spatial memory module tests passed!")
 
         finally:
             # Cleanup
             if "dimos" in locals():
-                dimos.close()
-
-
-if __name__ == "__main__":
-    pytest.main(["-v", "-s", __file__])
-    # test = TestSpatialMemoryModule()
-    # asyncio.run(
-    #     test.test_spatial_memory_module_with_replay(tempfile.mkdtemp(prefix="spatial_memory_test_"))
-    # )
+                dimos.close_all()
diff --git a/dimos/protocol/pubsub/impl/test_rospubsub.py b/dimos/protocol/pubsub/impl/test_rospubsub.py
index 6cf49c37b2..9add4ef893 100644
--- a/dimos/protocol/pubsub/impl/test_rospubsub.py
+++ b/dimos/protocol/pubsub/impl/test_rospubsub.py
@@ -50,7 +50,6 @@ def subscriber() -> Generator[DimosROS, None, None]:
     yield from ros_node()
 
 
-@pytest.mark.ros
 def test_basic_conversion(publisher, subscriber):
     """Test Vector3 publish/subscribe through ROS.
 
@@ -76,7 +75,7 @@ def callback(msg, t):
     assert msg.z == 3.0
 
 
-@pytest.mark.ros
+@pytest.mark.slow
 def test_pointcloud2_pubsub(publisher, subscriber):
     """Test PointCloud2 publish/subscribe through ROS.
 
@@ -133,7 +132,6 @@ def callback(msg, t):
     assert abs(original.ts - converted.ts) < 0.001
 
 
-@pytest.mark.ros
 def test_pointcloud2_empty_pubsub(publisher, subscriber):
     """Test empty PointCloud2 publish/subscribe.
 
@@ -162,7 +160,6 @@ def callback(msg, t):
     assert len(received[0]) == 0
 
 
-@pytest.mark.ros
 def test_posestamped_pubsub(publisher, subscriber):
     """Test PoseStamped publish/subscribe through ROS.
 
@@ -203,7 +200,6 @@ def callback(msg, t):
     np.testing.assert_allclose(converted.orientation.w, original.orientation.w, rtol=1e-5)
 
 
-@pytest.mark.ros
 def test_pointstamped_pubsub(publisher, subscriber):
     """Test PointStamped publish/subscribe through ROS.
 
@@ -246,7 +242,6 @@ def callback(msg, t):
     assert converted.point.z == original.point.z
 
 
-@pytest.mark.ros
 def test_twist_pubsub(publisher, subscriber):
     """Test Twist publish/subscribe through ROS.
 
diff --git a/dimos/protocol/pubsub/test_spec.py b/dimos/protocol/pubsub/test_spec.py
index 26c1cf0357..f79145642a 100644
--- a/dimos/protocol/pubsub/test_spec.py
+++ b/dimos/protocol/pubsub/test_spec.py
@@ -319,7 +319,7 @@ async def consume_messages() -> None:
         assert received_messages == messages_to_send
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 @pytest.mark.parametrize("pubsub_context, topic, values", testdata)
 def test_high_volume_messages(
     pubsub_context: Callable[[], Any], topic: Any, values: list[Any]
diff --git a/dimos/protocol/rpc/test_spec.py b/dimos/protocol/rpc/test_spec.py
index c29db13703..b5189c04bf 100644
--- a/dimos/protocol/rpc/test_spec.py
+++ b/dimos/protocol/rpc/test_spec.py
@@ -293,7 +293,7 @@ def callback(val) -> None:
             unsub_server()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 @pytest.mark.parametrize("rpc_context, impl_name", testdata)
 def test_timeout(rpc_context, impl_name: str) -> None:
     """Test that RPC calls properly timeout."""
@@ -392,8 +392,3 @@ def make_call(a, b) -> None:
 
         finally:
             unsub()
-
-
-if __name__ == "__main__":
-    # Run tests for debugging
-    pytest.main([__file__, "-v"])
diff --git a/dimos/robot/drone/test_drone.py b/dimos/robot/drone/test_drone.py
index d9075beae3..7381359f5a 100644
--- a/dimos/robot/drone/test_drone.py
+++ b/dimos/robot/drone/test_drone.py
@@ -1029,7 +1029,3 @@ def test_velocity_from_bbox_center_error(self) -> None:
         self.assertGreater(vy, 0)
         # No vertical offset -> vx should be ~0
         self.assertAlmostEqual(vx, 0, places=1)
-
-
-if __name__ == "__main__":
-    unittest.main()
diff --git a/dimos/robot/test_all_blueprints.py b/dimos/robot/test_all_blueprints.py
index 16f657393b..6c2d000ca8 100644
--- a/dimos/robot/test_all_blueprints.py
+++ b/dimos/robot/test_all_blueprints.py
@@ -25,7 +25,7 @@
 }
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 @pytest.mark.parametrize("blueprint_name", all_blueprints.keys())
 def test_all_blueprints_are_valid(blueprint_name: str) -> None:
     """Test that all blueprints in all_blueprints are valid Blueprint instances."""
diff --git a/dimos/robot/unitree/testing/test_actors.py b/dimos/robot/unitree/testing/test_actors.py
index 9366092eb6..0fee2175fc 100644
--- a/dimos/robot/unitree/testing/test_actors.py
+++ b/dimos/robot/unitree/testing/test_actors.py
@@ -102,12 +102,6 @@ def test_mapper_start(dimos) -> None:
     print("start res", mapper.start().result())
 
 
-if __name__ == "__main__":
-    dimos = core.start(2)
-    test_basic(dimos)
-    test_mapper_start(dimos)
-
-
 @pytest.mark.tool
 def test_counter(dimos) -> None:
     counter = dimos.deploy(Counter)
diff --git a/dimos/types/test_weaklist.py b/dimos/types/test_weaklist.py
index 06f9f851ce..447b2fdd9a 100644
--- a/dimos/types/test_weaklist.py
+++ b/dimos/types/test_weaklist.py
@@ -54,7 +54,7 @@ def test_weaklist_basic_operations() -> None:
     assert SampleObject(4) not in wl
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_weaklist_auto_removal() -> None:
     """Test that objects are automatically removed when garbage collected."""
     wl = WeakList()
@@ -137,7 +137,7 @@ def test_weaklist_clear() -> None:
     assert obj1 not in wl
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_weaklist_iteration_during_modification() -> None:
     """Test that iteration works even if objects are deleted during iteration."""
     wl = WeakList()
diff --git a/dimos/utils/cli/lcmspy/test_lcmspy.py b/dimos/utils/cli/lcmspy/test_lcmspy.py
index 530f081f29..13e6306c10 100644
--- a/dimos/utils/cli/lcmspy/test_lcmspy.py
+++ b/dimos/utils/cli/lcmspy/test_lcmspy.py
@@ -20,28 +20,46 @@
 from dimos.utils.cli.lcmspy.lcmspy import GraphLCMSpy, GraphTopic, LCMSpy, Topic as TopicSpy
 
 
-@pytest.mark.lcm
-def test_spy_basic() -> None:
+@pytest.fixture
+def pickle_lcm():
     lcm = PickleLCM(autoconf=True)
     lcm.start()
+    yield lcm
+    lcm.stop()
 
-    lcmspy = LCMSpy(autoconf=True)
-    lcmspy.start()
 
+@pytest.fixture
+def lcmspy_instance():
+    spy = LCMSpy(autoconf=True)
+    spy.start()
+    yield spy
+    spy.stop()
+
+
+@pytest.fixture
+def graph_lcmspy_instance():
+    spy = GraphLCMSpy(autoconf=True, graph_log_window=0.1)
+    spy.start()
+    time.sleep(0.2)  # Wait for thread to start
+    yield spy
+    spy.stop()
+
+
+def test_spy_basic(pickle_lcm, lcmspy_instance) -> None:
     video_topic = Topic(topic="/video")
     odom_topic = Topic(topic="/odom")
 
     for i in range(5):
-        lcm.publish(video_topic, f"video frame {i}")
+        pickle_lcm.publish(video_topic, f"video frame {i}")
         time.sleep(0.1)
         if i % 2 == 0:
-            lcm.publish(odom_topic, f"odometry data {i / 2}")
+            pickle_lcm.publish(odom_topic, f"odometry data {i / 2}")
 
     # Wait a bit for messages to be processed
     time.sleep(0.5)
 
     # Test statistics for video topic
-    video_topic_spy = lcmspy.topic["/video"]
+    video_topic_spy = lcmspy_instance.topic["/video"]
     assert video_topic_spy is not None
 
     # Test frequency (should be around 10 Hz for 5 messages in ~0.5 seconds)
@@ -60,7 +78,7 @@ def test_spy_basic() -> None:
     print(f"Video topic average message size: {avg_size:.2f} bytes")
 
     # Test statistics for odom topic
-    odom_topic_spy = lcmspy.topic["/odom"]
+    odom_topic_spy = lcmspy_instance.topic["/odom"]
     assert odom_topic_spy is not None
 
     freq = odom_topic_spy.freq(1.0)
@@ -79,7 +97,6 @@ def test_spy_basic() -> None:
     print(f"Odom topic: {odom_topic_spy}")
 
 
-@pytest.mark.lcm
 def test_topic_statistics_direct() -> None:
     """Test Topic statistics directly without LCM"""
 
@@ -128,7 +145,6 @@ def test_topic_cleanup() -> None:
     assert topic.message_history[0][0] > time.time() - 10  # Recent message
 
 
-@pytest.mark.lcm
 def test_graph_topic_basic() -> None:
     """Test GraphTopic basic functionality"""
     topic = GraphTopic("/test_graph")
@@ -144,79 +160,59 @@ def test_graph_topic_basic() -> None:
     assert topic.bandwidth_history[0] > 0
 
 
-@pytest.mark.lcm
-def test_graph_lcmspy_basic() -> None:
+def test_graph_lcmspy_basic(graph_lcmspy_instance) -> None:
     """Test GraphLCMSpy basic functionality"""
-    spy = GraphLCMSpy(autoconf=True, graph_log_window=0.1)
-    spy.start()
-    time.sleep(0.2)  # Wait for thread to start
-
     # Simulate a message
-    spy.msg("/test", b"test data")
+    graph_lcmspy_instance.msg("/test", b"test data")
     time.sleep(0.2)  # Wait for graph update
 
     # Should create GraphTopic with history
-    topic = spy.topic["/test"]
+    topic = graph_lcmspy_instance.topic["/test"]
     assert isinstance(topic, GraphTopic)
     assert len(topic.freq_history) > 0
     assert len(topic.bandwidth_history) > 0
 
-    spy.stop()
-
 
-@pytest.mark.lcm
-def test_lcmspy_global_totals() -> None:
+def test_lcmspy_global_totals(lcmspy_instance) -> None:
     """Test that LCMSpy tracks global totals as a Topic itself"""
-    spy = LCMSpy(autoconf=True)
-    spy.start()
-
     # Send messages to different topics
-    spy.msg("/video", b"video frame data")
-    spy.msg("/odom", b"odometry data")
-    spy.msg("/imu", b"imu data")
+    lcmspy_instance.msg("/video", b"video frame data")
+    lcmspy_instance.msg("/odom", b"odometry data")
+    lcmspy_instance.msg("/imu", b"imu data")
 
     # Verify each test topic received exactly one message (ignore LCM discovery packets)
     for t in ("/video", "/odom", "/imu"):
-        assert len(spy.topic[t].message_history) == 1
+        assert len(lcmspy_instance.topic[t].message_history) == 1
 
     # Check global statistics
-    global_freq = spy.freq(1.0)
-    global_kbps = spy.kbps(1.0)
-    global_size = spy.size(1.0)
+    global_freq = lcmspy_instance.freq(1.0)
+    global_kbps = lcmspy_instance.kbps(1.0)
+    global_size = lcmspy_instance.size(1.0)
 
     assert global_freq > 0
     assert global_kbps > 0
     assert global_size > 0
 
     print(f"Global frequency: {global_freq:.2f} Hz")
-    print(f"Global bandwidth: {spy.kbps_hr(1.0)}")
+    print(f"Global bandwidth: {lcmspy_instance.kbps_hr(1.0)}")
     print(f"Global avg message size: {global_size:.0f} bytes")
 
-    spy.stop()
 
-
-@pytest.mark.lcm
-def test_graph_lcmspy_global_totals() -> None:
+def test_graph_lcmspy_global_totals(graph_lcmspy_instance) -> None:
     """Test that GraphLCMSpy tracks global totals with history"""
-    spy = GraphLCMSpy(autoconf=True, graph_log_window=0.1)
-    spy.start()
-    time.sleep(0.2)
-
     # Send messages
-    spy.msg("/video", b"video frame data")
-    spy.msg("/odom", b"odometry data")
+    graph_lcmspy_instance.msg("/video", b"video frame data")
+    graph_lcmspy_instance.msg("/odom", b"odometry data")
     time.sleep(0.2)  # Wait for graph update
 
     # Update global graphs
-    spy.update_graphs(1.0)
+    graph_lcmspy_instance.update_graphs(1.0)
 
     # Should have global history
-    assert len(spy.freq_history) == 1
-    assert len(spy.bandwidth_history) == 1
-    assert spy.freq_history[0] > 0
-    assert spy.bandwidth_history[0] > 0
-
-    print(f"Global frequency history: {spy.freq_history[0]:.2f} Hz")
-    print(f"Global bandwidth history: {spy.bandwidth_history[0]:.2f} kB/s")
+    assert len(graph_lcmspy_instance.freq_history) == 1
+    assert len(graph_lcmspy_instance.bandwidth_history) == 1
+    assert graph_lcmspy_instance.freq_history[0] > 0
+    assert graph_lcmspy_instance.bandwidth_history[0] > 0
 
-    spy.stop()
+    print(f"Global frequency history: {graph_lcmspy_instance.freq_history[0]:.2f} Hz")
+    print(f"Global bandwidth history: {graph_lcmspy_instance.bandwidth_history[0]:.2f} kB/s")
diff --git a/dimos/utils/docs/test_doclinks.py b/dimos/utils/docs/test_doclinks.py
index 968f465cef..7da6a6281b 100644
--- a/dimos/utils/docs/test_doclinks.py
+++ b/dimos/utils/docs/test_doclinks.py
@@ -773,7 +773,3 @@ def test_skips_mailto_links(self, file_index, doc_index):
         assert len(errors) == 0
         assert len(changes) == 0
         assert new_content == content
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])
diff --git a/dimos/utils/test_data.py b/dimos/utils/test_data.py
index e5be4307c7..e55c8b20f3 100644
--- a/dimos/utils/test_data.py
+++ b/dimos/utils/test_data.py
@@ -23,7 +23,7 @@
 from dimos.utils.data import LfsPath
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 def test_pull_file() -> None:
     repo_root = data._get_repo_root()
     test_file_name = "cafe.jpg"
@@ -79,7 +79,7 @@ def test_pull_file() -> None:
         )
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 def test_pull_dir() -> None:
     repo_root = data._get_repo_root()
     test_dir_name = "ab_lidar_frames"
@@ -187,7 +187,7 @@ def test_lfs_path_no_download_on_creation() -> None:
     assert cache is None
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 def test_lfs_path_with_real_file() -> None:
     """Test LfsPath with a real small LFS file."""
     # Use a small existing LFS file
@@ -221,7 +221,7 @@ def test_lfs_path_with_real_file() -> None:
     assert content.startswith(b"\x89PNG")
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 def test_lfs_path_unload_and_reload() -> None:
     """Test unloading and reloading an LFS file."""
     filename = "three_paths.png"
@@ -266,7 +266,7 @@ def test_lfs_path_unload_and_reload() -> None:
     assert content_first == content_second
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 def test_lfs_path_operations() -> None:
     """Test various Path operations with LfsPath."""
     filename = "three_paths.png"
@@ -295,7 +295,7 @@ def test_lfs_path_operations() -> None:
     assert filename in fspath_result
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 def test_lfs_path_division_operator() -> None:
     """Test path division operator with LfsPath."""
     # Use a directory for testing
@@ -309,7 +309,7 @@ def test_lfs_path_division_operator() -> None:
     assert "three_paths.png" in str(result)
 
 
-@pytest.mark.heavy
+@pytest.mark.slow
 def test_lfs_path_multiple_instances() -> None:
     """Test that multiple LfsPath instances for same file work correctly."""
     filename = "three_paths.png"
diff --git a/dimos/utils/test_reactive.py b/dimos/utils/test_reactive.py
index 5bfc0a590f..f6f1340059 100644
--- a/dimos/utils/test_reactive.py
+++ b/dimos/utils/test_reactive.py
@@ -82,7 +82,7 @@ def _dispose() -> None:
     return proxy
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_backpressure_handling() -> None:
     # Create a dedicated scheduler for this test to avoid thread leaks
     test_scheduler = ThreadPoolScheduler(max_workers=8)
@@ -142,7 +142,7 @@ def test_backpressure_handling() -> None:
         test_scheduler.executor.shutdown(wait=True)
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_getter_streaming_blocking() -> None:
     source = dispose_spy(
         rx.interval(0.2).pipe(ops.map(lambda i: np.array([i, i + 1, i + 2])), ops.take(50))
@@ -177,7 +177,7 @@ def test_getter_streaming_blocking_timeout() -> None:
     assert source.is_disposed()
 
 
-@pytest.mark.integration
+@pytest.mark.slow
 def test_getter_streaming_nonblocking() -> None:
     source = dispose_spy(rx.interval(0.2).pipe(ops.take(50)))
 
diff --git a/dimos/utils/test_transform_utils.py b/dimos/utils/test_transform_utils.py
index b404579598..7923124c9f 100644
--- a/dimos/utils/test_transform_utils.py
+++ b/dimos/utils/test_transform_utils.py
@@ -672,7 +672,3 @@ def test_retract_arbitrary_pose(self) -> None:
         assert np.isclose(retracted.position.x, expected_x)
         assert np.isclose(retracted.position.y, expected_y)
         assert np.isclose(retracted.position.z, expected_z)
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])
diff --git a/docs/capabilities/manipulation/readme.md b/docs/capabilities/manipulation/readme.md
index 4a943e6be5..0d6539b75c 100644
--- a/docs/capabilities/manipulation/readme.md
+++ b/docs/capabilities/manipulation/readme.md
@@ -101,7 +101,7 @@ KeyboardTeleopModule ──→ ControlCoordinator ──→ ManipulationModule
 
 ## Adding a Custom Arm
 
-[guide is here](adding_a_custom_arm.md)
+[guide is here](/docs/capabilities/manipulation/adding_a_custom_arm.md)
 
 ## Key Files
 
diff --git a/docs/development/testing.md b/docs/development/testing.md
index c27a8c5dec..c8a226b7ad 100644
--- a/docs/development/testing.md
+++ b/docs/development/testing.md
@@ -8,24 +8,28 @@ uv sync --all-extras --no-extra dds
 
 ## Types of tests
 
-There are different types of tests based on what their goal is:
+In general, there are different types of tests based on what their goal is:
 
 | Type | Description | Mocking | Speed |
 |------|-------------|---------|-------|
-| Unit | Test a small individual piece of code | All dependencies | Very fast |
-| Integration | Test the integration between multiple units of code | Most dependencies | Some fast, some slow |
-| Functional | Test a particular desired functionality | Some dependencies | Some fast, some slow |
+| Unit | Test a small individual piece of code | All external systems | Very fast |
+| Integration | Test the integration between multiple units of code | Most external systems | Some fast, some slow |
+| Functional | Test a particular desired functionality | Some external systems | Some fast, some slow |
 | End-to-end | Test the entire system as a whole from the perspective of the user | None | Very slow |
 
 The distinction between unit, integration, and functional tests is often debated and rarely productive.
 
 Rather than waste time on classifying tests, it's better to separate tests by how they are used:
 
-* **fast tests**: tests which you can run after each code change (people often run them with filesystem watchers: whenever a file is saved, automatically run the tests)
-* **slow tests**: tests which you run every once in a while to make sure you haven't broken anything (maybe every commit, but definitely before publishing a PR)
+| Test Group | When to run | Typical usage |
+|------------|-------------|---------------|
+| **fast tests** | after each code change | often run with filesystem watchers so tests rerun whenever a file is saved |
+| **slow tests** | every once in a while to make sure you haven't broken anything | maybe every commit, but definitely before publishing a PR |
 
 The purpose of running tests in a loop is to get immediate feedback. The faster the loop, the easier it is to identify a problem since the source is the tiny bit of code you changed.
 
+For the purposes of DimOS, slow tests are marked with `@pytest.mark.slow` and fast tests are all the remaining ones.
+
 ## Usage
 
 ### Fast tests
@@ -42,7 +46,7 @@ This is the same as:
 pytest dimos
 ```
 
-The default `addopts` in `pyproject.toml` includes a `-m` filter that excludes slow markers (like `integration`, `heavy`, `e2e`, etc.), so plain `pytest dimos` only runs fast tests.
+The default `addopts` in `pyproject.toml` includes a `-m` filter that excludes the `slow`/`mujoco`/`tool`. So plain `pytest dimos` only runs fast tests.
 
 ### Slow tests
 
@@ -52,14 +56,14 @@ Run the slow tests:
 ./bin/pytest-slow
 ```
 
-This overrides the default `-m` filter to include most markers. When writing or debugging a specific slow test, override `-m` yourself:
+(This is just a shortcut for `pytest -m 'not (tool or mujoco)' dimos`. I.e., run both fast tests and slow tests, but not `tool` or `mujoco`.)
+
+When writing or debugging a specific slow test, override `-m` yourself to run it:
 
 ```bash
-pytest -m integration dimos/path/to/test_something.py
+pytest -m slow dimos/path/to/test_something.py
 ```
 
-Note: passing `-m` on the command line overrides the default from `addopts`, so you get exactly the marker set you asked for.
-
 ## Writing tests
 
 Test files live next to the code they test. If you have `dimos/core/pubsub.py`, its tests go in `dimos/core/test_pubsub.py`.
@@ -120,3 +124,17 @@ There are other useful things in `mocker`, like `mocker.MagicMock()` for creatin
 | `--pdb` | Drop into the debugger when a test fails |
 | `--tb=short` | Shorter tracebacks |
 | `--durations=0` | Measure the speed of each test |
+
+## Markers
+
+We have a few markers in use now.
+
+* `slow`: used to mark tests that take more than 1 second to finish.
+* `tool`: tests which require human interaction. I don't like this. Please don't use them.
+* `mujoco`: tests which use `MuJoCo`. These are very slow and don't work in CI currently.
+
+If a test needs to be skipped for some reason, please use on of these markers, or add another one.
+
+* `skipif_in_ci`: tests which cannot run in GitHub Actions
+* `skipif_no_openai`: tests which require an `OPENAI_API_KEY` key in the env
+* `skipif_no_alibaba`: tests which require an `ALIBABA_API_KEY` key in the env
diff --git a/docs/platforms/humanoid/g1/index.md b/docs/platforms/humanoid/g1/index.md
index 2e04f3b023..797c865b20 100644
--- a/docs/platforms/humanoid/g1/index.md
+++ b/docs/platforms/humanoid/g1/index.md
@@ -13,9 +13,9 @@ The Unitree G1 is a humanoid robot platform with full-body locomotion, arm gestu
 ## Install
 
 First, install system dependencies for your platform:
-- [Ubuntu](../../../installation/ubuntu.md)
-- [macOS](../../../installation/osx.md)
-- [Nix](../../../installation/nix.md)
+- [Ubuntu](/docs/installation/ubuntu.md)
+- [macOS](/docs/installation/osx.md)
+- [Nix](/docs/installation/nix.md)
 
 Then install DimOS:
 
@@ -159,9 +159,9 @@ primitive (sensors + vis)
 
 ## Deep Dive
 
-- [Navigation Stack](../../../capabilities/navigation/readme.md) — path planning and autonomous exploration
-- [Visualization](../../../usage/visualization.md) — Rerun, Foxglove, performance tuning
-- [Data Streams](../../../usage/data_streams/) — RxPY streams, backpressure, quality filtering
-- [Transports](../../../usage/transports/index.md) — LCM, SHM, DDS
-- [Blueprints](../../../usage/blueprints.md) — composing modules
-- [Agents](../../../capabilities/agents/readme.md) — LLM agent framework
+- [Navigation Stack](/docs/capabilities/navigation/readme.md) — path planning and autonomous exploration
+- [Visualization](/docs/usage/visualization.md) — Rerun, Foxglove, performance tuning
+- [Data Streams](/docs/usage/data_streams) — RxPY streams, backpressure, quality filtering
+- [Transports](/docs/usage/transports/index.md) — LCM, SHM, DDS
+- [Blueprints](/docs/usage/blueprints.md) — composing modules
+- [Agents](/docs/capabilities/agents/readme.md) — LLM agent framework
diff --git a/pyproject.toml b/pyproject.toml
index 6471fd89cd..f81f69887c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -388,24 +388,11 @@ follow_imports = "skip"
 
 [tool.pytest.ini_options]
 testpaths = ["dimos"]
-markers = [
-    "heavy: resource heavy test",
-    "tool: dev tooling",
-    "ros: depend on ros",
-    "lcm: tests that run actual LCM bus (can't execute in CI)",
-    "module: tests that need to run directly as modules",
-    "gpu: tests that require GPU",
-    "cuda: tests which require CUDA (specifically CUDA not just GPU acceleration)",
-    "e2e: end to end tests",
-    "integration: slower integration tests",
-    "neverending: they don't finish",
-    "mujoco: tests which open mujoco",
-]
 env = [
   "GOOGLE_MAPS_API_KEY=AIzafake_google_key",
   "PYTHONWARNINGS=ignore:cupyx.jit.rawkernel is experimental:FutureWarning",
 ]
-addopts = "-v -s -p no:warnings -ra --color=yes -m 'not (vis or exclude or tool or lcm or ros or heavy or gpu or module or e2e or integration or neverending or mujoco)'"
+addopts = "-v -r a -p no:warnings --color=yes -m 'not (tool or slow or mujoco)'"
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"