diff --git a/runpod/api/queries/apps.py b/runpod/api/queries/apps.py
index c5b11ac3..0d8b63cb 100644
--- a/runpod/api/queries/apps.py
+++ b/runpod/api/queries/apps.py
@@ -9,6 +9,27 @@
 }
 """
 
+QUERY_ENDPOINT_SNAPSHOT = """
+query EndpointSnapshot($id: String!) {
+    myself {
+        endpoint(id: $id) {
+            id name type instanceIds gpuIds gpuCount locations
+            workersMin workersMax workersPFBTarget idleTimeout
+            scalerType scalerValue executionTimeoutMs workerInitTimeoutMs
+            flashBootType minCudaVersion allowedCudaVersions
+            modelReferences env { key value }
+            networkVolumeIds { networkVolumeId }
+            volumes { volumeId volumeType }
+            template {
+                name imageName containerDiskInGb dockerArgs
+                env { key value }
+                ports containerRegistryAuthId advancedStart startScript config
+            }
+        }
+    }
+}
+"""
+
 QUERY_FLASH_APP_BY_NAME = """
 query getFlashAppByName($flashAppName: String!) {
     flashAppByName(flashAppName: $flashAppName) {
diff --git a/runpod/apps/api.py b/runpod/apps/api.py
index 49dd1d6c..f9d9882d 100644
--- a/runpod/apps/api.py
+++ b/runpod/apps/api.py
@@ -5,6 +5,7 @@
 absent from rest. management verbs for the wider sdk stay in runpod.api.ctl_commands.
 """
 
+from copy import deepcopy
 from typing import Any, Dict, List, Optional, Set
 
 import aiohttp
@@ -243,6 +244,39 @@ async def _execute(
         )
         return response["data"]
 
+    async def snapshot_endpoint(self, endpoint_id: str) -> Dict[str, Any]:
+        """capture a saveable endpoint configuration in memory for rollback."""
+        data = await self._execute(
+            app_queries.QUERY_ENDPOINT_SNAPSHOT, {"id": endpoint_id}, retry=True
+        )
+        endpoint = (data.get("myself") or {}).get("endpoint")
+        template = (endpoint or {}).get("template")
+        if (
+            not endpoint
+            or endpoint.get("id") != endpoint_id
+            or not template
+            or not template.get("imageName")
+        ):
+            raise RuntimeError(
+                f"endpoint {endpoint_id} has no restorable configuration"
+            )
+        snapshot = deepcopy(endpoint)
+        template = snapshot["template"]
+        snapshot["template"]["containerRegistryAuthId"] = (
+            template.get("containerRegistryAuthId") or ""
+        )
+        snapshot["template"]["dockerArgs"] = template.get("dockerArgs") or ""
+        snapshot["template"]["env"] = template.get("env") or []
+        snapshot["env"] = snapshot.get("env") or []
+        snapshot["modelReferences"] = snapshot.get("modelReferences") or []
+        snapshot["networkVolumeIds"] = snapshot.get("networkVolumeIds") or []
+        snapshot["volumes"] = [
+            volume
+            for volume in snapshot.get("volumes") or []
+            if volume["volumeType"] == "OBJECT_STORE_VOLUME"
+        ]
+        return snapshot
+
     async def save_endpoint(self, endpoint_input: Dict[str, Any]) -> Dict[str, Any]:
         """save via rest unless the input requires an unsupported capability."""
         body = _rest_endpoint_input(endpoint_input)
diff --git a/runpod/apps/deploy.py b/runpod/apps/deploy.py
index c166248b..9f94d3ce 100644
--- a/runpod/apps/deploy.py
+++ b/runpod/apps/deploy.py
@@ -1,15 +1,19 @@
 """deploy pipeline: discovered apps -> manifest -> artifact -> activated build."""
 
+import asyncio
 import json
 import logging
 import os
 import secrets
+import shlex
 import tarfile
 import tempfile
 from dataclasses import dataclass, field
 from pathlib import Path, PurePosixPath
 from typing import Any, Dict, List, Optional, Set
 
+import aiohttp
+
 from .api import AppsApiClient
 from .app import App
 from .build import (
@@ -19,7 +23,11 @@
     BuildResult,
     build_environment,
 )
-from .errors import InvalidResourceError, ScheduleNotSupported
+from .errors import (
+    EndpointReconciliationError,
+    InvalidResourceError,
+    ScheduleNotSupported,
+)
 from .discovery import resolve_project, validate_apps
 from .schedule import SCHEDULES_ENABLED
 from .shim import runtime_launcher
@@ -397,19 +405,18 @@ async def reconcile_endpoints(
     python_version: str,
     *,
     resolver=None,
+    activate_build: bool = False,
+    environment_created: bool = False,
+    events=None,
 ) -> Dict[str, str]:
-    """converge the environment's endpoints to the app's resources.
+    """snapshot and reconcile endpoints, restoring configurations on failure.
 
-    resource name -> endpoint id for everything provisioned. existing
-    endpoints (matched by name) are updated in place; endpoints whose
-    resources disappeared are deleted.
+    obsolete endpoints are removed after the deployment commits.
     """
     for handle in app.resources.values():
         handle.spec.validate()
     existing_endpoints = environment.get("endpoints") or []
     existing = {e["name"]: e["id"] for e in existing_endpoints}
-    # updates send the bound template's current name, so redeploying an
-    # existing endpoint sends the same template name as before.
     existing_templates = {
         e["name"]: (e.get("template") or {}).get("name") for e in existing_endpoints
     }
@@ -426,47 +433,179 @@ async def reconcile_endpoints(
 
     resolver = resolver or VolumeResolver(client)
     endpoints: Dict[str, str] = {}
-    for name, handle in sorted(provisionable.items()):
-        members = None
-        if needs_gpu_catalog(handle.spec.gpu):
-            members = await client.gpu_pool_members()
-        if name in existing:
-            template_name = existing_templates.get(name)
-        else:
-            template_name = _new_template_name(app, handle.spec, template_token)
-        payload = _deployed_endpoint_input(
-            app,
-            handle.spec,
-            environment["id"],
-            build_id,
-            python_version,
-            env_name=environment["name"],
-            pool_members=members or {},
-            template_name=template_name,
-        )
-        await attach_endpoint_volumes(payload, handle.spec, resolver, app)
-        auth_id = await resolve_registry_auth(handle.spec.registry_auth, api=client)
-        if auth_id:
-            payload["template"]["containerRegistryAuthId"] = auth_id
-        if handle.spec.model:
-            from .model import model_reference
-
-            payload["modelReferences"] = [model_reference(handle.spec.model)]
-        if name in existing:
-            payload["id"] = existing[name]
-        result = await client.save_endpoint(payload)
-        endpoints[name] = result["id"]
-        log.info(
-            "%s endpoint %s (%s)",
-            "updated" if name in existing else "provisioned",
-            name,
-            result["id"],
-        )
-
-    for name, endpoint_id in existing.items():
-        if name not in provisionable:
-            await client.delete_endpoint(endpoint_id)
-            log.info("deleted removed endpoint %s (%s)", name, endpoint_id)
+    snapshots = {}
+    undo = []
+    activation_attempted = False
+    creating = False
+    action = "prepare deployment"
+    try:
+        if provisionable.keys() & existing.keys():
+            _phase(events, "snapshot")
+        for name in sorted(provisionable.keys() & existing.keys()):
+            action = f"read '{name}'"
+            snapshots[name] = await client.snapshot_endpoint(existing[name])
+            snapshots[name]["flashEnvironmentId"] = environment["id"]
+        if activate_build:
+            action = "activate the build"
+            _phase(events, "activate")
+            activation_attempted = True
+            await client.deploy_build(environment["id"], build_id)
+        _phase(events, "endpoints")
+        for name, handle in sorted(provisionable.items()):
+            action = f"prepare '{name}'"
+            creating = False
+            members = None
+            if needs_gpu_catalog(handle.spec.gpu):
+                members = await client.gpu_pool_members()
+            if name in existing:
+                template_name = existing_templates.get(name)
+            else:
+                template_name = _new_template_name(app, handle.spec, template_token)
+            payload = _deployed_endpoint_input(
+                app,
+                handle.spec,
+                environment["id"],
+                build_id,
+                python_version,
+                env_name=environment["name"],
+                pool_members=members or {},
+                template_name=template_name,
+            )
+            await attach_endpoint_volumes(payload, handle.spec, resolver, app)
+            auth_id = await resolve_registry_auth(handle.spec.registry_auth, api=client)
+            if auth_id:
+                payload["template"]["containerRegistryAuthId"] = auth_id
+            if handle.spec.model:
+                from .model import model_reference
+
+                payload["modelReferences"] = [model_reference(handle.spec.model)]
+            if name in existing:
+                payload["id"] = existing[name]
+                undo.append((name, existing[name], snapshots[name]))
+            action = f"deploy '{name}'"
+            creating = name not in existing
+            result = await client.save_endpoint(payload)
+            endpoints[name] = result["id"]
+            if creating:
+                undo.append((name, result["id"], None))
+            creating = False
+            log.info("saved endpoint %s (%s)", name, result["id"])
+
+    except Exception as exc:
+        lines = [f"Couldn't {action}: {str(exc) or type(exc).__name__}"]
+        rollback_errors = []
+        rollback_allowed = True
+        if activation_attempted:
+            try:
+                current = await client.get_environment_by_name(
+                    app.name, environment["name"]
+                )
+                if not isinstance(current, dict) or "activeBuildId" not in current:
+                    raise RuntimeError("activation status unknown")
+                if current["activeBuildId"] != build_id and (
+                    undo or current["activeBuildId"] != environment.get("activeBuildId")
+                ):
+                    raise RuntimeError("environment changed during deployment")
+            except Exception as ownership_exc:
+                rollback_errors.append(
+                    f"build: {str(ownership_exc) or type(ownership_exc).__name__}"
+                )
+                rollback_allowed = False
+        if rollback_allowed and activation_attempted and not environment_created:
+            try:
+                previous_build = environment.get("activeBuildId")
+                if previous_build:
+                    await client.deploy_build(environment["id"], previous_build)
+                else:
+                    current = await client.get_environment_by_name(
+                        app.name, environment["name"]
+                    )
+                    if not isinstance(current, dict) or "activeBuildId" not in current:
+                        raise RuntimeError("activation status unknown")
+                    if current["activeBuildId"]:
+                        raise RuntimeError("no previous build to restore")
+            except Exception as restore_exc:
+                rollback_errors.append(
+                    f"build: {str(restore_exc) or type(restore_exc).__name__}"
+                )
+        if (
+            rollback_allowed
+            and creating
+            and isinstance(exc, (aiohttp.ClientError, OSError, asyncio.TimeoutError))
+        ):
+            try:
+                current = await client.get_environment_by_name(
+                    app.name, environment["name"]
+                )
+                matches = [
+                    endpoint
+                    for endpoint in (current or {}).get("endpoints", [])
+                    if (endpoint.get("template") or {}).get("name") == template_name
+                    and endpoint["id"] not in existing.values()
+                ]
+                if len(matches) != 1:
+                    raise RuntimeError("creation status unknown")
+                undo.append((name, matches[0]["id"], None))
+            except Exception:
+                rollback_errors.append(f"{name}: creation status unknown")
+        for resource, endpoint_id, snapshot in (
+            reversed(undo) if rollback_allowed else []
+        ):
+            try:
+                if snapshot is not None:
+                    await client.save_endpoint(snapshot)
+                elif not await client.delete_endpoint(endpoint_id):
+                    raise RuntimeError("endpoint deletion was not confirmed")
+            except Exception as cleanup_exc:
+                rollback_errors.append(
+                    f"{resource} ({endpoint_id}): {str(cleanup_exc) or type(cleanup_exc).__name__}"
+                )
+        if environment_created and not rollback_errors:
+            try:
+                current = await client.get_environment_by_name(
+                    app.name, environment["name"]
+                )
+                if (
+                    not isinstance(current, dict)
+                    or "endpoints" not in current
+                    or current["endpoints"]
+                ):
+                    raise RuntimeError("couldn't confirm the environment is empty")
+                if current.get("activeBuildId") not in (None, build_id):
+                    raise RuntimeError("environment changed during deployment")
+                if not await client.delete_environment(environment["id"]):
+                    raise RuntimeError("environment deletion was not confirmed")
+            except Exception as cleanup_exc:
+                rollback_errors.append(
+                    f"environment {environment['id']}: {str(cleanup_exc) or type(cleanup_exc).__name__}"
+                )
+        if rollback_errors:
+            lines.extend(
+                ["Rollback incomplete:", *(f"  {error}" for error in rollback_errors)]
+            )
+            lines.append(
+                f"Check `rp flash env get {shlex.quote(environment['name'])} "
+                f"--app {shlex.quote(app.name)}`. Remaining workers may still incur charges."
+            )
+        elif undo or activation_attempted:
+            lines.append("Changes rolled back.")
+        raise EndpointReconciliationError("\n".join(lines)) from exc
+
+    obsolete = existing.keys() - provisionable.keys()
+    if obsolete:
+        _phase(events, "cleanup")
+    for name in sorted(obsolete):
+        endpoint_id = existing[name]
+        try:
+            if not await client.delete_endpoint(endpoint_id):
+                raise RuntimeError("endpoint deletion was not confirmed")
+        except Exception as exc:
+            raise EndpointReconciliationError(
+                f"Deployed, but couldn't remove '{name}' ({endpoint_id}): {exc}\n"
+                f"Check `rp flash env get {shlex.quote(environment['name'])} "
+                f"--app {shlex.quote(app.name)}`."
+            ) from exc
+        log.info("deleted removed endpoint %s (%s)", name, endpoint_id)
 
     return endpoints
 
@@ -573,6 +712,7 @@ async def deploy_app(
 
     environments = {e["name"]: e for e in remote_app.get("flashEnvironments") or []}
     environment = environments.get(env_name)
+    environment_created = environment is None
     if environment is None:
         environment = await client.create_environment(app_id, env_name)
         log.info("created environment %s (%s)", env_name, environment["id"])
@@ -598,12 +738,10 @@ async def deploy_app(
         )
     log.info("uploaded build %s", build["id"])
 
-    await client.deploy_build(environment["id"], build["id"])
-    log.info("activated build %s on %s/%s", build["id"], app.name, env_name)
-
-    _phase(events, "endpoints")
     endpoints = await reconcile_endpoints(
-        client, app, environment, build["id"], python_version, resolver=resolver
+        client, app, environment, build["id"], python_version,
+        resolver=resolver, activate_build=True,
+        environment_created=environment_created, events=events,
     )
     endpoint_ready = getattr(events, "endpoint_ready", None)
     if endpoint_ready is not None:
diff --git a/runpod/apps/discovery.py b/runpod/apps/discovery.py
index 6211c985..efc5ce85 100644
--- a/runpod/apps/discovery.py
+++ b/runpod/apps/discovery.py
@@ -125,6 +125,9 @@ def _import_module(path: Path, project: Project) -> None:
                 "or standard-library module."
             )
     except (Exception, SystemExit) as exc:
+        if isinstance(exc, InvalidResourceError):
+            relative = path.relative_to(project.source_root)
+            raise DiscoveryError(f"{relative}: {exc}") from exc
         hint = ""
         if isinstance(exc, ModuleNotFoundError) and (
             project.source_root != project.import_root
diff --git a/runpod/apps/errors.py b/runpod/apps/errors.py
index 2d34177f..78a12ddb 100644
--- a/runpod/apps/errors.py
+++ b/runpod/apps/errors.py
@@ -49,3 +49,7 @@ def __init__(self) -> None:
 
 class InvalidResourceError(AppError):
     """a decorator was applied to an unsupported target or with bad config."""
+
+
+class EndpointReconciliationError(AppError):
+    """endpoint reconciliation failed with rollback results and remaining changes."""
diff --git a/runpod/apps/source.py b/runpod/apps/source.py
index 2130198b..e35bc80a 100644
--- a/runpod/apps/source.py
+++ b/runpod/apps/source.py
@@ -131,7 +131,7 @@ def iter_source_files(
                 )
             ):
                 if not discovery:
-                    log.warning("excluded source path %r from deployment artifact", rel)
+                    log.debug("excluded source path %r from deployment artifact", rel)
                 continue
             kept_dirs.append(name)
         dirs[:] = kept_dirs
@@ -147,6 +147,6 @@ def iter_source_files(
                 or (discovery and (path.suffix != ".py" or name == "setup.py"))
             ):
                 if not discovery:
-                    log.warning("excluded source path %r from deployment artifact", rel)
+                    log.debug("excluded source path %r from deployment artifact", rel)
                 continue
             yield path
diff --git a/runpod/apps/spec.py b/runpod/apps/spec.py
index 391471e7..26d03391 100644
--- a/runpod/apps/spec.py
+++ b/runpod/apps/spec.py
@@ -126,6 +126,13 @@ def normalize_cpu(
         )
     if not cpu or any(not isinstance(c, str) or not c.strip() for c in cpu):
         raise InvalidResourceError("cpu must contain non-empty instance ids")
+    from ..api.ctl_commands import _cpu_config
+
+    for instance_id in cpu:
+        try:
+            _cpu_config(instance_id)
+        except ValueError as exc:
+            raise InvalidResourceError(f"unknown cpu {instance_id!r}. {exc}") from exc
     return list(cpu)
 
 
@@ -246,6 +253,14 @@ def validate(self) -> None:
             raise InvalidResourceError(
                 f"resource '{self.name}': gpu_count must be >= 1"
             )
+        if (
+            isinstance(self.idle_timeout, bool)
+            or not isinstance(self.idle_timeout, int)
+            or not 1 <= self.idle_timeout <= 3600
+        ):
+            raise InvalidResourceError(
+                f"'{self.name}': idle_timeout must be a whole number from 1 to 3600 seconds"
+            )
         if self.max_concurrency < 1:
             raise InvalidResourceError(
                 f"resource '{self.name}': max_concurrency must be >= 1, "
diff --git a/runpod/rp_cli/console.py b/runpod/rp_cli/console.py
index 3bb10b7b..3cb24fb6 100644
--- a/runpod/rp_cli/console.py
+++ b/runpod/rp_cli/console.py
@@ -14,11 +14,13 @@
 from rich.progress import (
     Progress,
     SpinnerColumn,
+    Task,
     TaskID,
     TextColumn,
 )
 from rich.spinner import SPINNERS
 from rich.theme import Theme
+from rich.text import Text
 
 from runpod._hosts import console_url
 
@@ -258,7 +260,10 @@ def reload_banner(module: str = "") -> None:
     "vendor": "resolving dependencies",
     "package": "packaging artifact",
     "upload": "uploading build",
-    "endpoints": "reconciling endpoints",
+    "endpoints": "deploying endpoints",
+    "snapshot": "saving current settings",
+    "activate": "activating build",
+    "cleanup": "removing unused endpoints",
 }
 
 
@@ -275,6 +280,13 @@ def _bar(fraction: float) -> str:
     )
 
 
+class _DeploySpinnerColumn(SpinnerColumn):
+    def render(self, task: Task) -> Text:
+        if task.fields.get("failed"):
+            return Text("✗", style="err")
+        return super().render(task)
+
+
 class DeployEvents:
     """deploy_app event sink: animated phase list with live progress.
 
@@ -285,7 +297,9 @@ class DeployEvents:
 
     def __init__(self) -> None:
         self._progress = Progress(
-            _spinner_column(),
+            _DeploySpinnerColumn(
+                RUNPOD_SAND_SPINNER, style=ACCENT, finished_text="[ok]✓[/ok]"
+            ),
             TextColumn("[white]{task.description}[/white]"),
             TextColumn("{task.fields[detail]}"),
             console=console,
@@ -358,8 +372,13 @@ def endpoint_ready(self, name: str, endpoint_id: str) -> None:
         self.endpoints: Dict[str, str] = getattr(self, "endpoints", {})
         self.endpoints[name] = endpoint_id
 
-    def close(self) -> None:
-        self._finish_current()
+    def close(self, *, failed: bool = False) -> None:
+        if failed and self._current is not None:
+            self._progress.update(self._current, failed=True, detail="")
+            self._progress.stop_task(self._current)
+            self._current = None
+        else:
+            self._finish_current()
         if self._started:
             self._progress.stop()
             self._started = False
diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py
index eae26a80..d8f1ad1a 100644
--- a/runpod/rp_cli/main.py
+++ b/runpod/rp_cli/main.py
@@ -225,6 +225,7 @@ def deploy(target, env, python_version, exclude, build_only):
         for found in apps:
             ui.set_name_width(list(found.resources))
             events = ui.DeployEvents()
+            succeeded = False
             try:
                 artifact = build_artifact(
                     found,
@@ -234,10 +235,11 @@ def deploy(target, env, python_version, exclude, build_only):
                     events=events,
                     output=Path.cwd() / f"{found.name}-artifact.tar.gz",
                 )
+                succeeded = True
             except Exception as exc:  # noqa: BLE001 - surface engine errors cleanly
                 raise click.ClickException(str(exc)) from exc
             finally:
-                events.close()
+                events.close(failed=not succeeded)
             size_mb = artifact.stat().st_size / (1024 * 1024)
             ui.success(
                 f"built [white]{found.name}[/white] "
@@ -268,6 +270,7 @@ def deploy(target, env, python_version, exclude, build_only):
             source=_app_source(found, project_root),
         )
         events = ui.DeployEvents()
+        succeeded = False
         with ui.Timer() as t:
             try:
                 result = asyncio.run(
@@ -280,10 +283,11 @@ def deploy(target, env, python_version, exclude, build_only):
                         events=events,
                     )
                 )
+                succeeded = True
             except Exception as exc:  # noqa: BLE001 - surface engine errors cleanly
                 raise click.ClickException(str(exc)) from exc
             finally:
-                events.close()
+                events.close(failed=not succeeded)
         ui.success(
             f"[bold white]{result.app_name}/{env or found.env}[/bold white] "
             f"is live [dim]{t.elapsed:.1f}s[/dim]"
diff --git a/tests/test_apps/test_deploy.py b/tests/test_apps/test_deploy.py
index 50d9187e..824e756f 100644
--- a/tests/test_apps/test_deploy.py
+++ b/tests/test_apps/test_deploy.py
@@ -196,6 +196,7 @@ def test_credential_like_source_paths_are_included(self, tmp_path):
                 assert f"src/{name}" in tar.getnames()
 
     def test_excluded_paths_are_logged_without_contents(self, tmp_path, caplog):
+        caplog.set_level("DEBUG", logger="runpod.apps.source")
         for name in (".env", "private.key", "private.pem", "local.txt", ".aws/config"):
             path = tmp_path / name
             path.parent.mkdir(parents=True, exist_ok=True)
@@ -210,6 +211,14 @@ def test_excluded_paths_are_logged_without_contents(self, tmp_path, caplog):
             )
         assert "sensitive-file-contents" not in caplog.text
 
+    def test_excluded_paths_are_quiet_at_normal_log_levels(self, tmp_path, caplog):
+        (tmp_path / "__pycache__").mkdir()
+        (tmp_path / ".env").write_text("sensitive-file-contents")
+        with tarfile.open(package_project(tmp_path, {})) as tar:
+            assert set(tar.getnames()) == {"runpod_manifest.json"}
+        assert "excluded source path" not in caplog.text
+        assert "sensitive-file-contents" not in caplog.text
+
     def test_vendored_env_included_under_env(self, tmp_path):
         _write_project(tmp_path)
         env_dir = tmp_path / "built-env"
diff --git a/tests/test_apps/test_deploy_event_failure.py b/tests/test_apps/test_deploy_event_failure.py
new file mode 100644
index 00000000..2b3e379c
--- /dev/null
+++ b/tests/test_apps/test_deploy_event_failure.py
@@ -0,0 +1,45 @@
+"""reconciliation errors before endpoint operations begin."""
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from runpod.apps import App
+from runpod.apps.app import _clear_registry
+from runpod.apps.deploy import reconcile_endpoints
+from runpod.apps.errors import EndpointReconciliationError
+
+
+@pytest.fixture(autouse=True)
+def clean_registry():
+    _clear_registry()
+    yield
+    _clear_registry()
+
+
+@pytest.mark.parametrize("existing", [False, True])
+async def test_phase_lookup_failure_preserves_original_error(existing):
+    error = RuntimeError("phase handler unavailable")
+
+    class Events:
+        @property
+        def phase(self):
+            raise error
+
+    app = App("demo")
+    app.queue(name="worker", cpu="cpu3c-1-2")(lambda: None)
+    api = AsyncMock()
+    environment = {
+        "id": "env",
+        "name": "prod",
+        "endpoints": [{"id": "ep-old", "name": "worker"}] if existing else [],
+    }
+
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(
+            api, app, environment, "new-build", "3.12", events=Events()
+        )
+
+    assert str(caught.value) == "Couldn't prepare deployment: phase handler unavailable"
+    assert caught.value.__cause__ is error
+    assert api.mock_calls == []
diff --git a/tests/test_apps/test_deploy_review_feedback.py b/tests/test_apps/test_deploy_review_feedback.py
new file mode 100644
index 00000000..9bc48119
--- /dev/null
+++ b/tests/test_apps/test_deploy_review_feedback.py
@@ -0,0 +1,113 @@
+"""cpu compatibility and deployment rollback safety."""
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from runpod.apps import App
+from runpod.apps.app import _clear_registry
+from runpod.apps.deploy import reconcile_endpoints
+from runpod.apps.errors import EndpointReconciliationError, InvalidResourceError
+
+
+@pytest.fixture(autouse=True)
+def clean_registry():
+    _clear_registry()
+    yield
+    _clear_registry()
+
+
+@pytest.mark.parametrize("kind", ["queue", "api", "task"])
+@pytest.mark.parametrize("cpu", ["cpu5g-4-16", "cpu3c-16-32", "cpu9z-32-64"])
+def test_cpu_instance_shapes_are_not_limited_to_enum_members(kind, cpu):
+    app = App("demo")
+    handle = getattr(app, kind)(name="worker", cpu=cpu)(lambda: None)
+    assert handle.spec.cpu == [cpu]
+
+
+@pytest.mark.parametrize(
+    "cpu",
+    ["cpu5g", "cpu5g-0-16", "cpu5g-4-0", "cpu5g-4", "cpu5g-4-16-extra", " cpu5g-4-16"],
+)
+def test_malformed_cpu_instance_ids_fail_before_deployment(cpu):
+    app = App("demo")
+    with pytest.raises(InvalidResourceError, match="unknown cpu"):
+        app.queue(name="worker", cpu=cpu)(lambda: None)
+
+
+async def test_asyncio_timeout_recovers_created_endpoint():
+    app = App("demo")
+    app.queue(name="worker", cpu="cpu3c-1-2")(lambda: None)
+    api = AsyncMock()
+    error = asyncio.TimeoutError("response lost")
+    api.save_endpoint.side_effect = error
+    api.get_environment_by_name.return_value = {
+        "endpoints": [
+            {
+                "id": "ep-created",
+                "name": "worker",
+                "template": {"name": "demo-worker-template-token"},
+            }
+        ]
+    }
+    with patch("runpod.apps.deploy.secrets.token_hex", return_value="token"):
+        with pytest.raises(EndpointReconciliationError) as caught:
+            await reconcile_endpoints(
+                api, app, {"id": "env", "name": "prod"}, "new-build", "3.12"
+            )
+    assert caught.value.__cause__ is error
+    assert (
+        str(caught.value)
+        == "Couldn't deploy 'worker': response lost\nChanges rolled back."
+    )
+    api.delete_endpoint.assert_awaited_once_with("ep-created")
+
+
+@pytest.mark.parametrize("environment_created", [False, True])
+@pytest.mark.parametrize("current", [{"activeBuildId": "other-build"}, {}, None])
+async def test_unconfirmed_ownership_prevents_rollback_writes(
+    current, environment_created
+):
+    app = App("demo")
+    for name in ("alpha", "beta", "gamma"):
+        app.queue(name=name, cpu="cpu3c-1-2")(lambda: None)
+    api = AsyncMock()
+    api.snapshot_endpoint.return_value = {"id": "ep-alpha", "name": "alpha"}
+    api.save_endpoint.side_effect = [
+        {"id": "ep-alpha"},
+        {"id": "ep-beta"},
+        RuntimeError("quota exceeded"),
+    ]
+    api.get_environment_by_name.return_value = current
+    environment = {
+        "id": "env",
+        "name": "prod",
+        "activeBuildId": None if environment_created else "old-build",
+        "endpoints": (
+            [] if environment_created else [{"id": "ep-alpha", "name": "alpha"}]
+        ),
+    }
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(
+            api,
+            app,
+            environment,
+            "new-build",
+            "3.12",
+            activate_build=True,
+            environment_created=environment_created,
+        )
+    message = str(caught.value)
+    assert "Rollback incomplete:" in message
+    assert "Changes rolled back." not in message
+    assert "rp flash env get prod --app demo" in message
+    assert "may still incur charges" in message
+    if current:
+        assert "environment changed during deployment" in message
+    else:
+        assert "activation status unknown" in message
+    api.deploy_build.assert_awaited_once_with("env", "new-build")
+    assert api.save_endpoint.await_count == 3
+    api.delete_endpoint.assert_not_awaited()
+    api.delete_environment.assert_not_awaited()
diff --git a/tests/test_apps/test_deploy_rollback.py b/tests/test_apps/test_deploy_rollback.py
new file mode 100644
index 00000000..b8a9fcbb
--- /dev/null
+++ b/tests/test_apps/test_deploy_rollback.py
@@ -0,0 +1,356 @@
+"""endpoint snapshots and deployment rollback ordering."""
+
+from copy import deepcopy
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from runpod.api.queries.apps import QUERY_ENDPOINT_SNAPSHOT
+from runpod.api.mutations.apps import MUTATION_SAVE_ENDPOINT
+from runpod.apps import App
+from runpod.apps.api import AppsApiClient
+from runpod.apps.app import _clear_registry
+from runpod.apps.deploy import reconcile_endpoints
+from runpod.apps.errors import EndpointReconciliationError
+
+
+@pytest.fixture(autouse=True)
+def clean_registry():
+    _clear_registry()
+    yield
+    _clear_registry()
+
+
+def snapshot(name="worker", endpoint_id="ep-old"):
+    return {
+        "id": endpoint_id,
+        "name": name,
+        "type": "QB",
+        "instanceIds": ["cpu3c-1-2"],
+        "gpuIds": None,
+        "gpuCount": 1,
+        "locations": "EU-RO-1",
+        "workersMin": 1,
+        "workersMax": 2,
+        "workersPFBTarget": 0,
+        "idleTimeout": 120,
+        "scalerType": "QUEUE_DELAY",
+        "scalerValue": 5,
+        "executionTimeoutMs": 5000,
+        "workerInitTimeoutMs": 10000,
+        "flashBootType": "OFF",
+        "minCudaVersion": None,
+        "allowedCudaVersions": "",
+        "modelReferences": [],
+        "env": [{"key": "CUSTOM", "value": "synthetic-private-value"}],
+        "networkVolumeIds": [{"networkVolumeId": "volume-old"}],
+        "volumes": [{"volumeId": "volume-old", "volumeType": "NETWORK_VOLUME"}],
+        "template": {
+            "name": "old-template",
+            "imageName": "old-image",
+            "containerDiskInGb": 10,
+            "dockerArgs": "old-command",
+            "env": [{"key": "RUNPOD_BUILD_ID", "value": "old-build"}],
+            "ports": None,
+            "containerRegistryAuthId": None,
+            "advancedStart": False,
+            "startScript": None,
+            "config": None,
+        },
+    }
+
+
+async def test_snapshot_is_a_restorable_copy_of_endpoint_and_template():
+    original = snapshot()
+    client = AppsApiClient()
+    client._execute = AsyncMock(return_value={"myself": {"endpoint": original}})
+    saved = await client.snapshot_endpoint("ep-old")
+    client._execute.assert_awaited_once_with(
+        QUERY_ENDPOINT_SNAPSHOT, {"id": "ep-old"}, retry=True
+    )
+    assert saved == {
+        **original,
+        "template": {**original["template"], "containerRegistryAuthId": ""},
+        "volumes": [],
+    }
+    saved["template"]["env"][0]["value"] = "modified"
+    assert original["template"]["env"][0]["value"] == "old-build"
+    assert original["template"]["containerRegistryAuthId"] is None
+
+
+async def test_snapshot_retains_global_volumes_and_empty_values():
+    original = snapshot()
+    original.update(env=None, modelReferences=None, networkVolumeIds=None)
+    original["template"].update(env=None, dockerArgs=None)
+    original["volumes"] = [
+        {"volumeId": "global-old", "volumeType": "OBJECT_STORE_VOLUME"}
+    ]
+    client = AppsApiClient()
+    client._execute = AsyncMock(return_value={"myself": {"endpoint": original}})
+    saved = await client.snapshot_endpoint("ep-old")
+    assert saved["volumes"] == original["volumes"]
+    assert saved["env"] == saved["modelReferences"] == saved["networkVolumeIds"] == []
+    assert saved["template"]["env"] == []
+    assert saved["template"]["dockerArgs"] == ""
+
+
+async def test_snapshot_can_be_saved_through_the_control_plane_client():
+    client = AppsApiClient()
+    client._execute = AsyncMock(
+        side_effect=[
+            {"myself": {"endpoint": snapshot()}},
+            {"saveEndpoint": {"id": "ep-old"}},
+        ]
+    )
+    saved = await client.snapshot_endpoint("ep-old")
+    assert await client.save_endpoint(saved) == {"id": "ep-old"}
+    client._execute.assert_awaited_with(MUTATION_SAVE_ENDPOINT, {"input": saved})
+
+
+@pytest.mark.parametrize(
+    "current", [None, {"activeBuildId": "new-build"}, {"activeBuildId": None}]
+)
+async def test_first_activation_failure_reports_unrestorable_build_state(current):
+    app = App("demo")
+    app.queue(name="worker", cpu="cpu3c-1-2")(lambda: None)
+    api = AsyncMock()
+    api.deploy_build.side_effect = TimeoutError("response lost")
+    api.get_environment_by_name.return_value = current
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(
+            api,
+            app,
+            {"id": "env", "name": "prod"},
+            "new-build",
+            "3.12",
+            activate_build=True,
+        )
+    if current == {"activeBuildId": None}:
+        assert "Changes rolled back." in str(caught.value)
+    else:
+        assert "Rollback incomplete:\n  build:" in str(caught.value)
+    api.save_endpoint.assert_not_awaited()
+
+
+@pytest.mark.parametrize(
+    "endpoint", [None, {"id": "other"}, {"id": "ep-old", "template": None}]
+)
+async def test_snapshot_fails_closed_when_current_state_is_unavailable(endpoint):
+    client = AppsApiClient()
+    client._execute = AsyncMock(return_value={"myself": {"endpoint": endpoint}})
+    with pytest.raises(RuntimeError, match="no restorable configuration"):
+        await client.snapshot_endpoint("ep-old")
+
+
+def app_and_environment():
+    app = App("demo")
+    for name in ("alpha", "beta", "gamma"):
+        app.queue(name=name, cpu="cpu3c-1-2")(lambda x: x)
+    environment = {
+        "id": "env",
+        "name": "prod",
+        "activeBuildId": "old-build",
+        "endpoints": [
+            {"id": "ep-alpha", "name": "alpha"},
+            {"id": "ep-beta", "name": "beta"},
+        ],
+    }
+    return app, environment
+
+
+async def test_restores_full_state_of_successful_and_failed_updates_in_reverse_order():
+    app, environment = app_and_environment()
+    alpha = snapshot("alpha", "ep-alpha")
+    beta = snapshot("beta", "ep-beta")
+    api = AsyncMock()
+    api.get_environment_by_name.return_value = {"activeBuildId": "new-build"}
+    api.snapshot_endpoint.side_effect = [deepcopy(alpha), deepcopy(beta)]
+    api.save_endpoint.side_effect = [
+        {"id": "ep-alpha"},
+        TimeoutError("response lost"),
+        {"id": "ep-beta"},
+        {"id": "ep-alpha"},
+    ]
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(
+            api, app, environment, "new-build", "3.12", activate_build=True
+        )
+    assert (
+        str(caught.value)
+        == "Couldn't deploy 'beta': response lost\nChanges rolled back."
+    )
+    calls = api.save_endpoint.await_args_list
+    assert calls[2].args[0] == {**beta, "flashEnvironmentId": "env"}
+    assert calls[3].args[0] == {**alpha, "flashEnvironmentId": "env"}
+    assert [call[0] for call in api.mock_calls[:3]] == [
+        "snapshot_endpoint",
+        "snapshot_endpoint",
+        "deploy_build",
+    ]
+    assert [call.args for call in api.deploy_build.await_args_list] == [
+        ("env", "new-build"),
+        ("env", "old-build"),
+    ]
+    assert api.mock_calls[-3][0] == "deploy_build"
+    api.delete_endpoint.assert_not_awaited()
+    assert "synthetic-private-value" not in str(caught.value)
+
+
+async def test_snapshot_failure_stops_before_any_writes():
+    app, environment = app_and_environment()
+    api = AsyncMock()
+    api.snapshot_endpoint.side_effect = [snapshot(), RuntimeError("access denied")]
+    with pytest.raises(
+        EndpointReconciliationError, match="Couldn't read 'beta': access denied"
+    ):
+        await reconcile_endpoints(
+            api, app, environment, "new-build", "3.12", activate_build=True
+        )
+    api.save_endpoint.assert_not_awaited()
+    api.deploy_build.assert_not_awaited()
+    api.delete_endpoint.assert_not_awaited()
+
+
+@pytest.mark.parametrize("restore_failure", [False, True])
+async def test_activation_failure_restores_endpoints_and_previous_build(
+    restore_failure,
+):
+    app, environment = app_and_environment()
+    api = AsyncMock()
+    api.get_environment_by_name.return_value = {"activeBuildId": "new-build"}
+    api.snapshot_endpoint.side_effect = [
+        snapshot("alpha", "ep-alpha"),
+        snapshot("beta", "ep-beta"),
+    ]
+    api.save_endpoint.side_effect = lambda payload: {
+        "id": payload.get("id", "ep-gamma")
+    }
+    api.deploy_build.side_effect = [
+        TimeoutError("activation timed out"),
+        RuntimeError("build restore denied") if restore_failure else {"id": "env"},
+    ]
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(
+            api, app, environment, "new-build", "3.12", activate_build=True
+        )
+    assert [call.args for call in api.deploy_build.await_args_list] == [
+        ("env", "new-build"),
+        ("env", "old-build"),
+    ]
+    api.delete_endpoint.assert_not_awaited()
+    api.save_endpoint.assert_not_awaited()
+    if restore_failure:
+        assert "Rollback incomplete:\n  build: build restore denied" in str(
+            caught.value
+        )
+    else:
+        assert (
+            str(caught.value)
+            == "Couldn't activate the build: activation timed out\nChanges rolled back."
+        )
+
+
+async def test_restore_failure_does_not_skip_other_endpoints():
+    app, environment = app_and_environment()
+    api = AsyncMock()
+    api.snapshot_endpoint.side_effect = [
+        snapshot("alpha", "ep-alpha"),
+        snapshot("beta", "ep-beta"),
+    ]
+    api.save_endpoint.side_effect = [
+        {"id": "ep-alpha"},
+        RuntimeError("update failed"),
+        RuntimeError("restore denied"),
+        {"id": "ep-alpha"},
+    ]
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(api, app, environment, "new-build", "3.12")
+    assert "Rollback incomplete:\n  beta (ep-beta): restore denied" in str(caught.value)
+    assert api.save_endpoint.await_args.args[0]["id"] == "ep-alpha"
+
+
+async def test_lost_create_response_is_recovered_by_unique_template_name():
+    app = App("demo")
+    app.queue(name="worker", cpu="cpu3c-1-2")(lambda: None)
+    environment = {"id": "env", "name": "prod", "endpoints": []}
+    api = AsyncMock()
+    api.save_endpoint.side_effect = TimeoutError("response lost")
+    api.get_environment_by_name.return_value = {
+        "endpoints": [
+            {
+                "id": "ep-created",
+                "name": "worker",
+                "template": {"name": "demo-worker-template-token"},
+            },
+            {
+                "id": "ep-other",
+                "name": "worker",
+                "template": {"name": "another-deploy"},
+            },
+        ]
+    }
+    with patch("runpod.apps.deploy.secrets.token_hex", return_value="token"):
+        with pytest.raises(EndpointReconciliationError) as caught:
+            await reconcile_endpoints(api, app, environment, "new-build", "3.12")
+    api.delete_endpoint.assert_awaited_once_with("ep-created")
+    assert (
+        str(caught.value)
+        == "Couldn't deploy 'worker': response lost\nChanges rolled back."
+    )
+
+
+async def test_activation_precedes_irreversible_cleanup():
+    app, environment = app_and_environment()
+    environment["endpoints"].append({"name": "obsolete", "id": "ep-obsolete"})
+    api = AsyncMock()
+    api.save_endpoint.side_effect = lambda payload: {
+        "id": payload.get("id", "ep-gamma")
+    }
+    await reconcile_endpoints(
+        api, app, environment, "new-build", "3.12", activate_build=True
+    )
+    calls = [call[0] for call in api.mock_calls]
+    assert calls.index("deploy_build") < min(
+        i for i, name in enumerate(calls) if name == "save_endpoint"
+    )
+    assert calls.index("delete_endpoint") > max(
+        i for i, name in enumerate(calls) if name == "save_endpoint"
+    )
+
+
+@pytest.mark.parametrize("cleanup_fails", [False, True])
+async def test_new_environment_is_removed_after_endpoint_rollback(cleanup_fails):
+    app = App("demo")
+    app.queue(name="alpha", cpu="cpu3c-1-2")(lambda: None)
+    app.queue(name="beta", cpu="cpu3c-1-2")(lambda: None)
+    environment = {"id": "env", "name": "prod", "endpoints": []}
+    api = AsyncMock()
+    api.save_endpoint.side_effect = [{"id": "ep-alpha"}, RuntimeError("quota exceeded")]
+    api.get_environment_by_name.return_value = {
+        "id": "env",
+        "endpoints": [],
+        "activeBuildId": "new-build",
+    }
+    if cleanup_fails:
+        api.delete_endpoint.side_effect = RuntimeError("delete denied")
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(
+            api,
+            app,
+            environment,
+            "new-build",
+            "3.12",
+            activate_build=True,
+            environment_created=True,
+        )
+    if cleanup_fails:
+        api.delete_environment.assert_not_awaited()
+        assert "Rollback incomplete:" in str(caught.value)
+    else:
+        api.delete_environment.assert_awaited_once_with("env")
+        calls = [call[0] for call in api.mock_calls]
+        assert calls.index("delete_endpoint") < calls.index("delete_environment")
+        assert (
+            str(caught.value)
+            == "Couldn't deploy 'beta': quota exceeded\nChanges rolled back."
+        )
diff --git a/tests/test_apps/test_partial_deploy.py b/tests/test_apps/test_partial_deploy.py
new file mode 100644
index 00000000..bc891637
--- /dev/null
+++ b/tests/test_apps/test_partial_deploy.py
@@ -0,0 +1,260 @@
+"""partial deployment diagnostics and preflight validation."""
+
+import io
+from unittest.mock import AsyncMock, Mock, patch
+
+import pytest
+from click.testing import CliRunner
+from rich.console import Console
+
+from runpod.apps import App
+from runpod.apps.app import _clear_registry
+from runpod.apps.build import BuildResult
+from runpod.apps.deploy import deploy_app, reconcile_endpoints
+from runpod.apps.errors import EndpointReconciliationError, InvalidResourceError
+from runpod.apps.gpu import CpuInstanceType
+from runpod.rp_cli import console as ui
+from runpod.rp_cli.main import cli
+
+
+@pytest.fixture(autouse=True)
+def clean_registry():
+    _clear_registry()
+    yield
+    _clear_registry()
+
+
+def make_app():
+    app = App("partial")
+    for name in ("c_later", "b_bad", "a_ok"):
+        app.queue(name=name, cpu="cpu3c-1-2")(lambda x: x)
+    return app
+
+
+def environment(endpoints=None):
+    return {
+        "id": "env-1",
+        "name": "prod",
+        "endpoints": endpoints or [],
+        "activeBuildId": "build-old",
+    }
+
+
+@pytest.mark.parametrize("updating", [False, True])
+@pytest.mark.parametrize(
+    "cause", [RuntimeError("quota exceeded"), TimeoutError("timed out")]
+)
+async def test_reports_confirmed_changes_and_failed_request(updating, cause):
+    api = AsyncMock()
+    snapshot = {"id": "ep-ok", "name": "a_ok", "workersMin": 2}
+    api.snapshot_endpoint.return_value = snapshot.copy()
+    api.get_environment_by_name.return_value = environment()
+    api.save_endpoint.side_effect = [{"id": "ep-ok"}, cause, {"id": "ep-ok"}]
+    existing = [{"name": "a_ok", "id": "ep-ok"}] if updating else []
+    existing.append({"name": "obsolete", "id": "ep-old"})
+
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(
+            api, make_app(), environment(existing), "build-1", "3.12"
+        )
+
+    message = str(caught.value)
+    assert caught.value.__cause__ is cause
+    assert f"Couldn't deploy 'b_bad': {cause}" in message
+    if isinstance(cause, TimeoutError):
+        assert "Rollback incomplete:\n  b_bad: creation status unknown" in message
+        assert "rp flash env get prod --app partial" in message
+    else:
+        assert (
+            message == "Couldn't deploy 'b_bad': quota exceeded\nChanges rolled back."
+        )
+    assert [call.args[0]["name"] for call in api.save_endpoint.await_args_list[:2]] == [
+        "a_ok",
+        "b_bad",
+    ]
+    if updating:
+        api.delete_endpoint.assert_not_awaited()
+        assert api.save_endpoint.await_args.args[0] == {
+            **snapshot,
+            "flashEnvironmentId": "env-1",
+        }
+    else:
+        api.delete_endpoint.assert_awaited_once_with("ep-ok")
+
+
+async def test_first_update_failure_identifies_existing_endpoint():
+    api = AsyncMock()
+    snapshot = {"id": "ep-old", "name": "a_ok", "workersMin": 2}
+    api.snapshot_endpoint.return_value = snapshot
+    api.save_endpoint.side_effect = [RuntimeError("rejected"), {"id": "ep-old"}]
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(
+            api,
+            make_app(),
+            environment([{"name": "a_ok", "id": "ep-old"}]),
+            "build-1",
+            "3.12",
+        )
+    assert str(caught.value) == "Couldn't deploy 'a_ok': rejected\nChanges rolled back."
+    api.delete_endpoint.assert_not_awaited()
+    assert api.save_endpoint.await_count == 2
+    assert api.save_endpoint.await_args.args[0] == snapshot
+
+
+async def test_preparation_failure_reports_earlier_changes():
+    api = AsyncMock()
+    api.save_endpoint.return_value = {"id": "ep-ok"}
+    with patch(
+        "runpod.apps.volume.attach_endpoint_volumes",
+        AsyncMock(side_effect=[None, RuntimeError("volume unavailable")]),
+    ):
+        with pytest.raises(EndpointReconciliationError) as caught:
+            await reconcile_endpoints(api, make_app(), environment(), "build-1", "3.12")
+    assert (
+        str(caught.value)
+        == "Couldn't prepare 'b_bad': volume unavailable\nChanges rolled back."
+    )
+    api.save_endpoint.assert_awaited_once()
+    api.delete_endpoint.assert_awaited_once_with("ep-ok")
+
+
+async def test_deletion_failure_reports_creates_updates_and_deletes():
+    api = AsyncMock()
+    api.save_endpoint.side_effect = lambda payload: {"id": f"ep-{payload['name']}"}
+    api.delete_endpoint.side_effect = [True, RuntimeError("delete denied"), True, True]
+    existing = [
+        {"name": name, "id": f"ep-{name}"} for name in ("b_bad", "old1", "old2", "old3")
+    ]
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(
+            api, make_app(), environment(existing), "build-1", "3.12"
+        )
+    message = str(caught.value)
+    assert "Deployed, but couldn't remove 'old2' (ep-old2): delete denied" in message
+    assert "rolled back" not in message
+    assert api.save_endpoint.await_count == 3
+    assert [call.args for call in api.delete_endpoint.await_args_list] == [
+        ("ep-old1",),
+        ("ep-old2",),
+    ]
+
+
+@pytest.mark.parametrize("cleanup_result", [RuntimeError("cleanup denied"), False])
+async def test_rollback_continues_after_cleanup_failure(cleanup_result):
+    api = AsyncMock()
+    cause = RuntimeError("quota exceeded")
+    api.save_endpoint.side_effect = [{"id": "ep-a"}, {"id": "ep-b"}, cause]
+    api.delete_endpoint.side_effect = [cleanup_result, True]
+    with pytest.raises(EndpointReconciliationError) as caught:
+        await reconcile_endpoints(api, make_app(), environment(), "build-1", "3.12")
+    message = str(caught.value)
+    assert caught.value.__cause__ is cause
+    assert "Couldn't deploy 'c_later': quota exceeded" in message
+    assert "Rollback incomplete:\n  b_bad (ep-b):" in message
+    assert "a_ok" not in message
+    assert "may still incur charges" in message
+    assert [call.args for call in api.delete_endpoint.await_args_list] == [
+        ("ep-b",),
+        ("ep-a",),
+    ]
+
+
+async def test_success_keeps_created_endpoints():
+    api = AsyncMock()
+    api.save_endpoint.side_effect = lambda payload: {"id": f"ep-{payload['name']}"}
+    result = await reconcile_endpoints(
+        api, make_app(), environment(), "build-1", "3.12"
+    )
+    assert result == {"a_ok": "ep-a_ok", "b_bad": "ep-b_bad", "c_later": "ep-c_later"}
+    api.delete_endpoint.assert_not_awaited()
+
+
+def test_cli_reports_partial_deploy_and_failed_phase(tmp_path, monkeypatch):
+    (tmp_path / "main.py").write_text(
+        'from runpod import App\napp = App("partial")\n'
+        '@app.queue(name="a_ok", cpu="cpu3c-1-2")\n'
+        "def a_ok(x): return x\n"
+        '@app.queue(name="b_bad", cpu="cpu3c-1-2")\n'
+        "def b_bad(x): return x\n"
+    )
+    env_dir = tmp_path / "vendor"
+    env_dir.mkdir()
+    api = AsyncMock()
+    api.get_app_by_name.return_value = {
+        "id": "app-1",
+        "flashEnvironments": [environment()],
+    }
+    api.prepare_artifact_upload.return_value = {
+        "uploadUrl": "https://upload.invalid",
+        "objectKey": "key",
+    }
+    api.finalize_artifact_upload.return_value = {"id": "build-1"}
+    api.get_environment_by_name.return_value = {"activeBuildId": "build-1"}
+    api.save_endpoint.side_effect = [{"id": "ep-ok"}, RuntimeError("quota exceeded")]
+    buffer = io.StringIO()
+    monkeypatch.setattr(ui, "console", Console(file=buffer, theme=ui.theme, width=120))
+    with (
+        patch("runpod.apps.deploy.AppsApiClient", return_value=api),
+        patch(
+            "runpod.apps.deploy.build_environment",
+            return_value=BuildResult(env_dir, []),
+        ),
+    ):
+        result = CliRunner().invoke(
+            cli, ["flash", "deploy", str(tmp_path), "-e", "prod"]
+        )
+    assert result.exit_code == 1, result.output
+    assert (
+        result.output
+        == "Error: Couldn't deploy 'b_bad': quota exceeded\nChanges rolled back.\n"
+    )
+    assert "✗ deploying endpoints" in buffer.getvalue()
+    assert "✓ deploying endpoints" not in buffer.getvalue()
+    assert "is live" not in buffer.getvalue()
+    assert [call.args for call in api.deploy_build.await_args_list] == [
+        ("env-1", "build-1"),
+        ("env-1", "build-old"),
+    ]
+    api.delete_endpoint.assert_awaited_once_with("ep-ok")
+
+
+@pytest.mark.parametrize("kind", ["queue", "api"])
+@pytest.mark.parametrize("timeout", [0, -1, 3601, 5000, 99999, 1.5, True, "60", None])
+def test_idle_timeout_rejected_at_decoration(kind, timeout):
+    app = App("invalid")
+    with pytest.raises(InvalidResourceError, match="idle_timeout"):
+        getattr(app, kind)(name="worker", cpu="cpu3c-1-2", idle_timeout=timeout)(
+            lambda: None
+        )
+
+
+@pytest.mark.parametrize("timeout", [1, 3600])
+@pytest.mark.parametrize("cpu", list(CpuInstanceType))
+def test_accepts_timeout_boundaries_and_supported_cpu_flavors(timeout, cpu):
+    app = App("valid")
+    handle = app.queue(name="worker", cpu=cpu, idle_timeout=timeout)(lambda: None)
+    assert handle.spec.idle_timeout == timeout
+    assert handle.spec.cpu == [cpu.value]
+
+
+@pytest.mark.parametrize("kind", ["queue", "api", "task"])
+@pytest.mark.parametrize("cpu", ["cpu9z", ["cpu3c-1-2", "cpu9z"]])
+def test_cpu_flavor_rejected_at_decoration(kind, cpu):
+    app = App("invalid")
+    with pytest.raises(InvalidResourceError, match="unknown cpu 'cpu9z'"):
+        getattr(app, kind)(name="worker", cpu=cpu)(lambda: None)
+
+
+@pytest.mark.parametrize("field,value", [("idle_timeout", 5000), ("cpu", ["cpu9z"])])
+async def test_mutated_invalid_spec_fails_before_build_and_remote_calls(
+    tmp_path, field, value
+):
+    app = make_app()
+    setattr(app.resources["b_bad"].spec, field, value)
+    api = AsyncMock()
+    build = Mock()
+    with patch("runpod.apps.deploy.build_environment", build):
+        with pytest.raises(InvalidResourceError):
+            await deploy_app(app, tmp_path, api=api)
+    build.assert_not_called()
+    assert not api.mock_calls