fix(deployment): join background deploy thread to stop cross-test state leak

src/runpod_flash/core/deployment.py

@@ -6,7 +6,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
-from typing import List
+from typing import List, Optional
 
 from rich.console import Console
 from rich.progress import Progress, SpinnerColumn, TextColumn
@@ -55,18 +55,27 @@ def __init__(self, max_concurrent: int = 3):
         self.manager = ResourceManager()
         self.results: List[DeploymentResult] = []
 
-    def deploy_all_background(self, resources: List[DeployableResource]) -> None:
+    def deploy_all_background(
+        self, resources: List[DeployableResource]
+    ) -> Optional[threading.Thread]:
         """Deploy all resources in background thread.
 
         This method spawns a background thread to deploy resources without
         blocking the main thread. Progress is logged to console.
 
         Args:
             resources: List of resources to deploy
+
+        Returns:
+            The worker thread, or None when there was nothing to deploy.
+            Callers that need the deployment to finish within a bounded scope
+            (notably tests, which patch the deploy path) must join it —
+            otherwise the daemon thread outlives that scope and mutates
+            ResourceManager's class-level state afterwards.
         """
         if not resources:
             console.print("[dim]No resources to deploy[/dim]")
-            return
+            return None
 
         def run_async_deployment():
             """Run async deployment in background thread."""
@@ -90,6 +99,7 @@ def run_async_deployment():
         console.print(
             f"[dim]Auto-provisioning {len(resources)} resource(s) in background...[/dim]"
         )
+        return thread
 
     async def deploy_all(
         self, resources: List[DeployableResource], show_progress: bool = True

tests/unit/test_deployment.py

@@ -2,6 +2,7 @@
 
 import pytest
 import asyncio
+import threading
 from unittest.mock import MagicMock, AsyncMock, patch
 
 from runpod_flash.core.deployment import (
@@ -204,17 +205,43 @@ def test_deploy_all_background(self, mock_resources):
             mock_deploy.side_effect = mock_resources
 
             # Should not block
-            orchestrator.deploy_all_background(mock_resources)
+            thread = orchestrator.deploy_all_background(mock_resources)
+
+            # Join inside the patch context so the worker cannot outlive the
+            # mock and reach the real deploy path. See the returns-a-thread
+            # test below for why this matters.
+            thread.join(timeout=10)
+            assert not thread.is_alive()
+
+    def test_deploy_all_background_returns_joinable_thread(self, mock_resources):
+        """Callers must be able to await background deployment.
+
+        Without a handle to join, the daemon thread outlives the caller's
+        patches and runs the real deploy path afterwards, writing mock
+        resources into the ResourceManager class-level state. That state is
+        shared process-wide, so an unrelated test later fails when it tries
+        to cloudpickle the leftover mock.
+        """
+        orchestrator = DeploymentOrchestrator()
+
+        with patch.object(
+            orchestrator.manager, "get_or_deploy_resource", new_callable=AsyncMock
+        ) as mock_deploy:
+            mock_deploy.side_effect = mock_resources
+
+            thread = orchestrator.deploy_all_background(mock_resources)
 
-            # Background thread should be started
-            # (not much we can test here without waiting for thread)
+            assert isinstance(thread, threading.Thread)
+            thread.join(timeout=10)
+            assert not thread.is_alive()
+            assert mock_deploy.await_count == len(mock_resources)
 
     def test_deploy_all_background_empty_list(self):
         """Test background deployment with empty list."""
         orchestrator = DeploymentOrchestrator()
 
-        # Should handle gracefully
-        orchestrator.deploy_all_background([])
+        # Nothing to deploy means no thread to join.
+        assert orchestrator.deploy_all_background([]) is None
 
     @pytest.mark.asyncio
     async def test_deploy_all_raises_api_key_error_before_deploying(