fix(app): tear down serverless endpoints on flash app delete

src/runpod_flash/cli/commands/apps.py

@@ -109,6 +109,38 @@ async def get_flash_app(app_name: str):
 
 
 async def delete_flash_app(app_name: str):
+    app = await FlashApp.from_name(app_name)
+
+    with console.status("[dim]deleting endpoints...[/dim]"):
+        removed, failed = await app.delete_endpoints()
+
+    for endpoint in removed:
+        label = endpoint["name"] or endpoint["id"]
+        console.print(
+            f"[green]\u2713[/green] deleted endpoint [bold]{label}[/bold]"
+            f"  [dim]{endpoint['id']}[/dim]"
+        )
+
+    if failed:
+        print_error(
+            console,
+            f"could not remove all endpoints for app '{app_name}'; app was not deleted",
+        )
+        for endpoint in failed:
+            endpoint_id = endpoint["id"]
+            if endpoint_id:
+                console.print(
+                    f"  endpoint {endpoint_id} not removed; "
+                    f"delete it with runpodctl serverless delete {endpoint_id}"
+                )
+            else:
+                name = endpoint["name"] or "(unnamed)"
+                console.print(
+                    f"  endpoint [bold]{name}[/bold] has no id in environment "
+                    f"'{endpoint['env']}'; remove it from the RunPod console"
+                )
+        raise typer.Exit(1)
+
     with console.status("[dim]deleting...[/dim]"):
         success = await FlashApp.delete(app_name=app_name)
     if success:

src/runpod_flash/core/api/runpod.py

@@ -986,3 +986,41 @@ async def __aenter__(self):
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self.close()
+
+
+async def _delete_endpoint_idempotent(
+    client: "RunpodGraphQLClient", endpoint_id: str
+) -> bool:
+    """Delete a serverless endpoint; treat already-deleted endpoints as removed.
+
+    If the delete call fails with an exception, the endpoint is re-checked and
+    a missing endpoint counts as removed, so retries of a partially-completed
+    teardown can proceed. A delete that returns success=False (without raising)
+    is a hard failure and is NOT re-checked.
+
+    Returns:
+        True if the endpoint was deleted (or already gone), False otherwise.
+    """
+    try:
+        result = await client.delete_endpoint(endpoint_id)
+        success = result.get("success", False)
+        if success:
+            log.debug(f"endpoint {endpoint_id} successfully deleted")
+            return True
+        log.debug(f"endpoint {endpoint_id} failed to delete")
+        return False
+
+    except Exception as e:
+        log.debug(f"endpoint {endpoint_id} failed to delete: {e}")
+
+        # Deletion failed. Check if endpoint still exists.
+        # If it doesn't exist, treat as successfully removed (already deleted).
+        try:
+            if not await client.endpoint_exists(endpoint_id):
+                log.debug(f"endpoint {endpoint_id} no longer exists on RunPod")
+                return True
+        except Exception as check_error:
+            log.warning(
+                f"Could not verify endpoint {endpoint_id} existence: {check_error}"
+            )
+        return False

src/runpod_flash/core/deployment.py

@@ -55,18 +55,25 @@ 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]
+    ) -> threading.Thread | None:
         """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 background thread running the deployment, or None when the
+            resource list is empty (no thread is started). Callers that need
+            deterministic completion (e.g. tests) may join() the thread.
         """
         if not resources:
             console.print("[dim]No resources to deploy[/dim]")
-            return
+            return None
 
         def run_async_deployment():
             """Run async deployment in background thread."""
@@ -91,6 +98,8 @@ def run_async_deployment():
             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
     ) -> List[DeploymentResult]:

src/runpod_flash/core/resources/app.py

@@ -4,7 +4,7 @@
 from typing import Dict, Optional, Union, Tuple, TYPE_CHECKING, Any, List
 import logging
 
-from ..api.runpod import RunpodGraphQLClient
+from ..api.runpod import RunpodGraphQLClient, _delete_endpoint_idempotent
 
 from .constants import (
     TARBALL_CONTENT_TYPE,
@@ -561,6 +561,55 @@ async def delete_environment(self, environment_name: str) -> bool:
             result = await client.delete_flash_environment(environment_id)
         return result.get("success", False)
 
+    async def delete_endpoints(
+        self,
+    ) -> Tuple[List[Dict[str, str]], List[Dict[str, str]]]:
+        """Delete all serverless endpoints registered to this app's environments.
+
+        Endpoints are discovered server-side from each environment record, so
+        this works without local resource tracking (e.g. in CI or from a
+        different machine than the deployer).
+
+        Safe to retry: endpoints that no longer exist remotely are treated as
+        already removed, so a partially-failed earlier attempt does not block
+        a subsequent run.
+
+        Returns:
+            Tuple of (removed, failed) endpoint descriptors. Each descriptor is
+            a dict with 'id', 'name', and 'env' keys.
+
+        Raises:
+            RuntimeError: If app is not hydrated (no ID available)
+        """
+        removed: List[Dict[str, str]] = []
+        failed: List[Dict[str, str]] = []
+
+        await self._hydrate()
+        async with RunpodGraphQLClient() as client:
+            environments = await client.list_flash_environments_by_app_id(self.id)
+            for environment in environments:
+                env_name = environment.get("name") or ""
+                env = await client.get_flash_environment(
+                    {"flashEnvironmentId": environment["id"]}
+                )
+                for endpoint in (env or {}).get("endpoints") or []:
+                    endpoint_id = endpoint.get("id")
+                    descriptor = {
+                        "id": endpoint_id or "",
+                        "name": endpoint.get("name") or "",
+                        "env": env_name,
+                    }
+                    if not endpoint_id:
+                        log.warning(
+                            "Environment '%s' lists an endpoint with no id", env_name
+                        )
+                        failed.append(descriptor)
+                        continue
+                    deleted = await _delete_endpoint_idempotent(client, endpoint_id)
+                    (removed if deleted else failed).append(descriptor)
+
+        return removed, failed
+
     async def list_builds(self) -> List[Dict[str, Any]]:
         """List all builds for this app.
 

src/runpod_flash/core/resources/serverless.py

@@ -19,7 +19,7 @@
 )
 from runpod.endpoint.runner import Job
 
-from ..api.runpod import RunpodGraphQLClient
+from ..api.runpod import RunpodGraphQLClient, _delete_endpoint_idempotent
 from ..exceptions import RunpodAPIKeyError
 from ..utils.backoff import get_backoff_delay
 from .base import DeployableResource
@@ -1340,34 +1340,8 @@ async def _do_undeploy(self) -> bool:
             log.warning(f"{self} has no endpoint ID, cannot undeploy")
             return False
 
-        try:
-            async with RunpodGraphQLClient() as client:
-                result = await client.delete_endpoint(self.id)
-                success = result.get("success", False)
-
-                if success:
-                    log.debug(f"{self} successfully undeployed")
-                    return True
-                else:
-                    log.debug(f"{self} failed to undeploy")
-                    return False
-
-        except Exception as e:
-            log.debug(f"{self} failed to undeploy: {e}")
-
-            # Deletion failed. Check if endpoint still exists.
-            # If it doesn't exist, treat as successful cleanup (orphaned endpoint).
-            try:
-                async with RunpodGraphQLClient() as client:
-                    if not await client.endpoint_exists(self.id):
-                        log.debug(
-                            f"{self} no longer exists on RunPod, removing from cache"
-                        )
-                        return True
-            except Exception as check_error:
-                log.warning(f"Could not verify endpoint existence: {check_error}")
-
-            return False
+        async with RunpodGraphQLClient() as client:
+            return await _delete_endpoint_idempotent(client, self.id)
 
     async def undeploy(self) -> Dict[str, Any]:
         resource_manager = ResourceManager()

tests/unit/cli/test_apps.py

@@ -194,10 +194,25 @@ def test_get_app_without_related_data(
 
 
 class TestAppsDelete:
+    @staticmethod
+    def _flash_app(removed=None, failed=None):
+        flash_app = MagicMock()
+        flash_app.delete_endpoints = AsyncMock(
+            return_value=(removed or [], failed or [])
+        )
+        return flash_app
+
+    @patch("runpod_flash.cli.commands.apps.FlashApp.from_name", new_callable=AsyncMock)
     @patch("runpod_flash.cli.commands.apps.FlashApp.delete", new_callable=AsyncMock)
     def test_delete_app_success(
-        self, mock_delete, runner, mock_asyncio_run_coro, patched_console
+        self,
+        mock_delete,
+        mock_from_name,
+        runner,
+        mock_asyncio_run_coro,
+        patched_console,
     ):
+        mock_from_name.return_value = self._flash_app()
         mock_delete.return_value = True
 
         with patch(
@@ -207,6 +222,7 @@ def test_delete_app_success(
             result = runner.invoke(app, ["app", "delete", "demo"])
 
         assert result.exit_code == 0
+        mock_from_name.assert_awaited_once_with("demo")
         mock_delete.assert_awaited_once_with(app_name="demo")
         printed = " ".join(
             str(call.args[0])
@@ -216,10 +232,106 @@ def test_delete_app_success(
         assert "deleted" in printed
         assert "demo" in printed
 
+    @patch("runpod_flash.cli.commands.apps.FlashApp.from_name", new_callable=AsyncMock)
+    @patch("runpod_flash.cli.commands.apps.FlashApp.delete", new_callable=AsyncMock)
+    def test_delete_app_removes_endpoints(
+        self,
+        mock_delete,
+        mock_from_name,
+        runner,
+        mock_asyncio_run_coro,
+        patched_console,
+    ):
+        removed = [{"id": "ep-1", "name": "api", "env": "dev"}]
+        mock_from_name.return_value = self._flash_app(removed=removed)
+        mock_delete.return_value = True
+
+        with patch(
+            "runpod_flash.cli.commands.apps.asyncio.run",
+            side_effect=mock_asyncio_run_coro,
+        ):
+            result = runner.invoke(app, ["app", "delete", "demo"])
+
+        assert result.exit_code == 0
+        mock_delete.assert_awaited_once_with(app_name="demo")
+        printed = " ".join(
+            str(call.args[0])
+            for call in patched_console.print.call_args_list
+            if call.args
+        )
+        assert "deleted endpoint" in printed
+        assert "api" in printed
+        assert "ep-1" in printed
+
+    @patch("runpod_flash.cli.commands.apps.FlashApp.from_name", new_callable=AsyncMock)
+    @patch("runpod_flash.cli.commands.apps.FlashApp.delete", new_callable=AsyncMock)
+    def test_delete_app_endpoint_failure_blocks_app_delete(
+        self,
+        mock_delete,
+        mock_from_name,
+        runner,
+        mock_asyncio_run_coro,
+        patched_console,
+    ):
+        failed = [{"id": "ep-fail", "name": "api", "env": "dev"}]
+        mock_from_name.return_value = self._flash_app(failed=failed)
+
+        with patch(
+            "runpod_flash.cli.commands.apps.asyncio.run",
+            side_effect=mock_asyncio_run_coro,
+        ):
+            result = runner.invoke(app, ["app", "delete", "demo"])
+
+        assert result.exit_code == 1
+        mock_delete.assert_not_awaited()
+        printed = " ".join(
+            str(call.args[0])
+            for call in patched_console.print.call_args_list
+            if call.args
+        )
+        assert "endpoint ep-fail not removed" in printed
+        assert "runpodctl serverless delete ep-fail" in printed
+
+    @patch("runpod_flash.cli.commands.apps.FlashApp.from_name", new_callable=AsyncMock)
+    @patch("runpod_flash.cli.commands.apps.FlashApp.delete", new_callable=AsyncMock)
+    def test_delete_app_endpoint_without_id_reports_console_remediation(
+        self,
+        mock_delete,
+        mock_from_name,
+        runner,
+        mock_asyncio_run_coro,
+        patched_console,
+    ):
+        failed = [{"id": "", "name": "mystery", "env": "dev"}]
+        mock_from_name.return_value = self._flash_app(failed=failed)
+
+        with patch(
+            "runpod_flash.cli.commands.apps.asyncio.run",
+            side_effect=mock_asyncio_run_coro,
+        ):
+            result = runner.invoke(app, ["app", "delete", "demo"])
+
+        assert result.exit_code == 1
+        mock_delete.assert_not_awaited()
+        printed = " ".join(
+            str(call.args[0])
+            for call in patched_console.print.call_args_list
+            if call.args
+        )
+        assert "mystery" in printed
+        assert "remove it from the RunPod console" in printed
+
+    @patch("runpod_flash.cli.commands.apps.FlashApp.from_name", new_callable=AsyncMock)
     @patch("runpod_flash.cli.commands.apps.FlashApp.delete", new_callable=AsyncMock)
     def test_delete_app_failure_raises_exit(
-        self, mock_delete, runner, mock_asyncio_run_coro, patched_console
+        self,
+        mock_delete,
+        mock_from_name,
+        runner,
+        mock_asyncio_run_coro,
+        patched_console,
     ):
+        mock_from_name.return_value = self._flash_app()
         mock_delete.return_value = False
 
         with patch(