fix(network-volume): re-resolve stale cached volume id by name

src/runpod_flash/core/resources/network_volume.py

@@ -108,8 +108,29 @@ def url(self) -> str:
     async def is_deployed(self) -> bool:
         """
         Checks if the network volume resource is deployed and available.
+
+        A cached id alone is not sufficient: the volume may have been deleted
+        on Runpod out of band, leaving a stale id in local state
+        (``.flash/resources.pkl``). Validate the id against the live API so a
+        stale id triggers re-resolution/recreation rather than a hard failure
+        downstream (see SLS-337). On a transient API error, treat the volume as
+        not deployed so deploy() falls back to resolve-by-name, which is
+        idempotent.
         """
-        return self.id is not None
+        if not self.id:
+            return False
+        try:
+            async with RunpodRestClient() as client:
+                return await self._volume_id_exists(client, self.id)
+        except Exception as e:
+            log.debug(f"Error checking {self}: {e}")
+            return False
+
+    async def _volume_id_exists(self, client, volume_id: str) -> bool:
+        """Return True if a volume with the given id still exists on Runpod."""
+        volumes_response = await client.list_network_volumes()
+        existing_volumes = self._normalize_volumes_response(volumes_response)
+        return any(volume.get("id") == volume_id for volume in existing_volumes)
 
     def _normalize_volumes_response(self, volumes_response) -> list:
         """Normalize API response to list format."""
@@ -163,17 +184,30 @@ async def _do_deploy(self) -> "DeployableResource":
         Returns a DeployableResource object.
         """
         try:
-            # If the resource is already deployed, return it
+            # If the resource is already deployed (id validated against the
+            # live API), return it. is_deployed() returns False for a stale
+            # cached id, so we fall through to re-resolve/recreate below.
             if await self.is_deployed():
                 log.debug(f"{self} exists")
                 return self
 
             async with RunpodRestClient() as client:
-                # Check for existing volume first
+                # Resolve by name first (source of truth). This also refreshes a
+                # stale cached id to the live one when the name still resolves.
                 if existing_volume := await self._find_existing_volume(client):
                     return existing_volume
 
-                # No existing volume found, create a new one
+                # Not found by name. Without a name there is nothing to
+                # re-resolve or safely recreate, so a stale id is terminal.
+                if not self.name:
+                    raise ValueError(
+                        f"Network volume id '{self.id}' no longer exists and "
+                        "cannot be re-resolved without a name."
+                    )
+
+                # Drop any stale cached id so it is not sent in the create
+                # payload, then create a fresh volume by name.
+                self.id = None
                 return await self._create_new_volume(client)
 
         except Exception as e:

tests/unit/resources/test_network_volume.py

@@ -198,6 +198,108 @@ def test_unknown_field_rejected(self):
         with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
             NetworkVolume(name="data", sizee=500)
 
+    @pytest.mark.asyncio
+    async def test_is_deployed_false_when_no_id(self):
+        """is_deployed() is False without an id and makes no API call."""
+        volume = NetworkVolume(name="deanq")
+
+        with patch(
+            "runpod_flash.core.resources.network_volume.RunpodRestClient"
+        ) as mock_client_class:
+            assert await volume.is_deployed() is False
+            mock_client_class.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_is_deployed_true_when_cached_id_exists(
+        self, mock_runpod_client, sample_volume_data
+    ):
+        """is_deployed() validates the cached id against the live API."""
+        volume = NetworkVolume(name="deanq")
+        volume.id = "vol-123456"
+        mock_runpod_client.list_network_volumes.return_value = [sample_volume_data]
+
+        with patch(
+            "runpod_flash.core.resources.network_volume.RunpodRestClient"
+        ) as mock_client_class:
+            mock_client_class.return_value.__aenter__.return_value = mock_runpod_client
+            mock_client_class.return_value.__aexit__ = AsyncMock()
+            assert await volume.is_deployed() is True
+
+    @pytest.mark.asyncio
+    async def test_is_deployed_false_when_cached_id_missing(self, mock_runpod_client):
+        """A stale cached id (deleted out of band) reads as not deployed."""
+        volume = NetworkVolume(name="deanq")
+        volume.id = "stale-id"
+        mock_runpod_client.list_network_volumes.return_value = []
+
+        with patch(
+            "runpod_flash.core.resources.network_volume.RunpodRestClient"
+        ) as mock_client_class:
+            mock_client_class.return_value.__aenter__.return_value = mock_runpod_client
+            mock_client_class.return_value.__aexit__ = AsyncMock()
+            assert await volume.is_deployed() is False
+
+    @pytest.mark.asyncio
+    async def test_do_deploy_recreates_by_name_when_cached_id_stale(
+        self, mock_runpod_client, sample_volume_data
+    ):
+        """Stale cached id falls back to resolve-by-name and create-if-missing."""
+        volume = NetworkVolume(name="deanq", size=50)
+        volume.id = "stale-id"
+
+        # Volume was deleted out of band: not present in the live list.
+        mock_runpod_client.list_network_volumes.return_value = []
+        mock_runpod_client.create_network_volume.return_value = sample_volume_data
+
+        with patch(
+            "runpod_flash.core.resources.network_volume.RunpodRestClient"
+        ) as mock_client_class:
+            mock_client_class.return_value.__aenter__.return_value = mock_runpod_client
+            mock_client_class.return_value.__aexit__ = AsyncMock()
+            result = await volume._do_deploy()
+
+        mock_runpod_client.create_network_volume.assert_called_once()
+        # The create payload must not carry the stale id forward.
+        payload = mock_runpod_client.create_network_volume.call_args.args[0]
+        assert "id" not in payload
+        assert result.id == "vol-123456"
+
+    @pytest.mark.asyncio
+    async def test_do_deploy_reuses_when_cached_id_valid(
+        self, mock_runpod_client, sample_volume_data
+    ):
+        """A valid cached id short-circuits without creating a new volume."""
+        volume = NetworkVolume(name="deanq", size=50)
+        volume.id = "vol-123456"
+        mock_runpod_client.list_network_volumes.return_value = [sample_volume_data]
+
+        with patch(
+            "runpod_flash.core.resources.network_volume.RunpodRestClient"
+        ) as mock_client_class:
+            mock_client_class.return_value.__aenter__.return_value = mock_runpod_client
+            mock_client_class.return_value.__aexit__ = AsyncMock()
+            result = await volume._do_deploy()
+
+        mock_runpod_client.create_network_volume.assert_not_called()
+        assert result.id == "vol-123456"
+
+    @pytest.mark.asyncio
+    async def test_do_deploy_raises_for_stale_id_only_volume(self, mock_runpod_client):
+        """An id-only volume that no longer exists cannot be re-resolved."""
+        volume = NetworkVolume(id="stale-id")
+        mock_runpod_client.list_network_volumes.return_value = []
+
+        with patch(
+            "runpod_flash.core.resources.network_volume.RunpodRestClient"
+        ) as mock_client_class:
+            mock_client_class.return_value.__aenter__.return_value = mock_runpod_client
+            # __aexit__ must not suppress the exception (return falsy).
+            mock_client_class.return_value.__aexit__ = AsyncMock(return_value=False)
+            with pytest.raises(ValueError, match="stale-id"):
+                await volume._do_deploy()
+
+        mock_runpod_client.create_network_volume.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_deploy_uses_resource_manager_to_register(self, sample_volume_data):
         """deploy() should go through the ResourceManager for persistence."""