Repository: skypilot-org/skypilot
Pull request #5581 (2025-05-14 07:28:28)

[Core] Add backward compatibility for instance naming in k8s
<!-- Describe the changes in this PR -->
Fix https://github.com/skypilot-org/skypilot/issues/5574
Add backward compatibility for instance naming in k8s

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [x] Code formatting: install pre-commit (auto-check on commit) or `bash format.sh`
- [x] Any manual or new tests for this PR (please specify below)
  - sky launch/exec with new clusters
  - sky launch/exec/status/queue/down with old clusters
- [ ] All smoke tests: `/smoke-test` (CI) or `pytest tests/test_smoke.py` (local)
- [ ] Relevant individual tests: `/smoke-test -k test_name` (CI) or `pytest tests/test_smoke.py::test_name` (local)
- [ ] Backward compatibility: `/quicktest-core` (CI) or `pytest tests/smoke_tests/test_backward_compat.py` (local)

<!-- CI commands (/-prefixed) can only be triggered by repo members -->



Patch:
diff --git a/.github/workflows/pypi-nightly-build.yml b/.github/workflows/pypi-nightly-build.yml
index ce0f38ace..6b2166fb5 100644
--- a/.github/workflows/pypi-nightly-build.yml
+++ b/.github/workflows/pypi-nightly-build.yml
@@ -26,6 +26,7 @@ jobs:
     if: ${{ needs.check-date.outputs.should_run != 'false' }}
     outputs:
       buildkite_json: ${{ steps.trigger_buildkite.outputs.json }}
+      version: ${{ steps.set-release-version.outputs.version }}
     steps:
       - name: Clone repository
         uses: actions/checkout@v3
@@ -55,11 +56,14 @@ jobs:
           build
           --user
       - name: Set release version
+        id: set-release-version
         run: |
           RELEASE_VERSION=$(date +%Y%m%d)
           sed -i "s/{{SKYPILOT_COMMIT_SHA}}/${{ github.sha }}/g" sky/__init__.py
           sed -i "s/__version__ = '.*'/__version__ = '1.0.0.dev${RELEASE_VERSION}'/g" sky/__init__.py
           sed -i "s/name='skypilot',/name='skypilot-nightly',/g" sky/setup_files/setup.py
+          echo "version=1.0.0.dev${RELEASE_VERSION}" >> $GITHUB_OUTPUT
+          echo "Version: $version"
       - name: Build a binary wheel and a source tarball
         run: >-
           python -m
@@ -105,7 +109,7 @@ jobs:
       BUILDKITE_TOKEN: ${{ secrets.BUILDKITE_TOKEN }}
 
   publish-to-pypi:
-    needs: wait-for-buildkite
+    needs: [nightly-build-pypi, wait-for-buildkite]
     if: needs.wait-for-buildkite.outputs.build_status == 'success'
     runs-on: ubuntu-latest
     steps:
@@ -122,6 +126,32 @@ jobs:
           repository-url: https://test.pypi.org/legacy/
           skip-existing: true
 
+      - name: Validate published package
+        run: |
+          # Wait a moment for package to be available
+          sleep 30
+          export SKYPILOT_DISABLE_USAGE_COLLECTION=1
+          pip install fastapi
+          # Install skypilot from Test PyPI with fallback to PyPI for dependencies
+          pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple skypilot-nightly
+          # Verify installation
+          sky -v
+          sky -c
+          sky check
+          sky show-gpus
+          sky dashboard
+          # Verify dashboard endpoint is accessible
+          curl --fail --silent --show-error --output /dev/null http://127.0.0.1:46580/dashboard/ || echo "Dashboard endpoint is not accessible" && exit 1
+          # Verify version
+          EXPECTED_VERSION="${{ needs.nightly-build-pypi.outputs.version }}"
+          INSTALLED_VERSION=$(pip show skypilot-nightly | grep Version | awk '{print $2}')
+          echo "Expected version: $EXPECTED_VERSION, Installed version: $INSTALLED_VERSION"
+          if [ "$INSTALLED_VERSION" != "$EXPECTED_VERSION" ]; then
+            echo "Version mismatch!"
+            exit 1
+          fi
+          echo "Version verified successfully!"
+
       - name: Publish distribution to PyPI
         uses: pypa/gh-action-pypi-publish@release/v1
         with:
diff --git a/.github/workflows/release-publish.yml b/.github/workflows/release-publish.yml
index e0c369760..ebd8db5ca 100644
--- a/.github/workflows/release-publish.yml
+++ b/.github/workflows/release-publish.yml
@@ -93,6 +93,9 @@ jobs:
           sky -c
           sky check
           sky show-gpus
+          sky dashboard
+          # Verify dashboard endpoint is accessible
+          curl --fail --silent --show-error --output /dev/null http://127.0.0.1:46580/dashboard/ || echo "Dashboard endpoint is not accessible" && exit 1
           # Verify version
           EXPECTED_VERSION="${{ needs.extract-version.outputs.version }}"
           INSTALLED_VERSION=$(pip show skypilot | grep Version | awk '{print $2}')
@@ -116,6 +119,15 @@ jobs:
       package_name: 'skypilot'
       version: ${{ needs.extract-version.outputs.version }}
     secrets: inherit
+  
+  publish-helm:
+    needs: [extract-version, publish-to-pypi, trigger-docker-build]
+    uses: ./.github/workflows/publish-helm.yml
+    with:
+      version: ${{ needs.extract-version.outputs.version }}
+      commit_message: 'Update Helm charts for skypilot version ${{ needs.extract-version.outputs.version }}'
+      package_name: 'skypilot'
+    secrets: inherit
 
   cleanup-branches:
     needs: extract-version
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index bf49d8088..e90717cf6 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -41,7 +41,7 @@ repos:
             # From tests/mypy_files.txt
             - "sky"
             - "--exclude"
-            - "sky/benchmark|sky/callbacks|sky/skylet/providers/azure|sky/resources.py|sky/backends/monkey_patches"
+            - "sky/benchmark|sky/callbacks|sky/backends/monkey_patches"
         pass_filenames: false
         additional_dependencies:
             - types-aiofiles
diff --git a/docs/requirements-docs.txt b/docs/requirements-docs.txt
index 6566296f7..8ad29b501 100644
--- a/docs/requirements-docs.txt
+++ b/docs/requirements-docs.txt
@@ -19,3 +19,4 @@ sphinxcontrib-htmlhelp==2.0.4
 sphinxcontrib-jsmath==1.0.1
 sphinxcontrib-qthelp==1.0.6
 sphinxcontrib-serializinghtml==1.1.9
+sphinx-tabs==3.4.7
diff --git a/docs/source/conf.py b/docs/source/conf.py
index bbfcb05f9..9799a522b 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -36,6 +36,7 @@ extensions = [
     'sphinx.ext.viewcode',
     'sphinx_autodoc_typehints',
     'sphinx_click',
+    'sphinx_tabs.tabs',
     'sphinx_copybutton',
     "sphinx_togglebutton",  # Used for collapsible sections in examples.
     'sphinxcontrib.googleanalytics',
diff --git a/docs/source/examples/frameworks/aws_efa.md b/docs/source/examples/frameworks/aws_efa.md
new file mode 120000
index 000000000..fe45a9043
--- /dev/null
+++ b/docs/source/examples/frameworks/aws_efa.md
@@ -0,0 +1,1 @@
+../../generated-examples/aws_efa.md
\ No newline at end of file
diff --git a/docs/source/examples/frameworks/gcp_gpu_direct_tcpx.md b/docs/source/examples/frameworks/gcp_gpu_direct_tcpx.md
new file mode 120000
index 000000000..cab896d31
--- /dev/null
+++ b/docs/source/examples/frameworks/gcp_gpu_direct_tcpx.md
@@ -0,0 +1,1 @@
+../../generated-examples/gcp_gpu_direct_tcpx.md
\ No newline at end of file
diff --git a/docs/source/examples/frameworks/index.rst b/docs/source/examples/frameworks/index.rst
index f8b0592ef..a1793959c 100644
--- a/docs/source/examples/frameworks/index.rst
+++ b/docs/source/examples/frameworks/index.rst
@@ -5,9 +5,11 @@ Frameworks
    :maxdepth: 1
 
    Airflow <airflow>
+   AWS EFA <aws_efa>
    Cross-cloud data transfer <https://nebius.com/blog/posts/bulk-object-storage-s3-data-migration-with-skypilot>
    DVC <dvc>
    GCP DWS/Kueue <https://gke-ai-labs.dev/docs/tutorials/skypilot/resource-management-using-kueue/>
+   GCP GPUDirect-TCPX <gcp_gpu_direct_tcpx>
    Jupyter <jupyter>
    MLFlow <https://nebius.com/blog/posts/orchestrating-llm-fine-tuning-k8s-skypilot-mlflow>
    MPI <mpi>
diff --git a/docs/source/examples/index.rst b/docs/source/examples/index.rst
index 39546f9fd..3ccfc3d2c 100644
--- a/docs/source/examples/index.rst
+++ b/docs/source/examples/index.rst
@@ -45,7 +45,7 @@ We welcome contributions from the community. Instructions below:
    4. Go to the appropriate subdir under Examples and add your example.
 
       - For example, if you want to add to "Examples / Serving":
-      - ``cd docs/source/examples/serving; ln -s ../generated-examples/my-llm.md``.
+      - ``cd docs/source/examples/serving; ln -s ../../generated-examples/my-llm.md``.
       - Add a heading for your example to the TOC in the index.rst file.
       - ``git add index.rst my-llm.md``
 
diff --git a/docs/source/reference/training-guide.rst b/docs/source/reference/training-guide.rst
index 0e0795ffe..b37ceaf64 100644
--- a/docs/source/reference/training-guide.rst
+++ b/docs/source/reference/training-guide.rst
@@ -37,6 +37,46 @@ Example configuration:
       A100-80GB:1  
       H100:1  
 
+Use high-performance networking
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. tabs::
+
+    .. tab:: AWS EFA
+
+        AWS Elastic Fabric Adapter (EFA) is a network interface similar to Nvidia Infiniband that enables users to run applications requiring high levels of inter-node communications at scale on AWS. You can enable EFA on AWS HyperPod/EKS clusters with a simple additional setting in your SkyPilot YAML.
+
+        Example configuration:
+
+        .. code-block:: yaml
+
+          config:
+            kubernetes:
+              pod_config:
+                spec:
+                  containers:
+                    - resources:
+                      limits:
+                        vpc.amazonaws.com/efa: 4
+                      requests:
+                        vpc.amazonaws.com/efa: 4
+
+        See `EFA example <https://github.com/skypilot-org/skypilot/tree/master/examples/aws_efa>`_ for more details.
+
+    .. tab:: GCP GPUDirect-TCPX
+
+        `GPUDirect-TCPX <https://cloud.google.com/compute/docs/gpus/gpudirect>`_ is a high-performance networking technology that enables direct communication between GPUs and network interfaces for `a3-highgpu-8g` or `a3-edgegpu-8g` VMs. You can enable it with the following additional setting in your SkyPilot YAML.
+
+        Example configuration:
+
+        .. code-block:: yaml
+
+          config:
+            gcp:
+              enable_gpu_direct: true
+
+        See `GPUDirect-TCPX example <https://github.com/skypilot-org/skypilot/tree/master/examples/gcp_gpu_direct_tcpx>`_ for more details.
+
 Use ``disk_tier: best``
 ~~~~~~~~~~~~~~~~~~~~~~~
 Fast storage is critical for loading and storing data and model checkpoints.
diff --git a/examples/aws_efa/README.md b/examples/aws_efa/README.md
index f342b1326..4813227c3 100644
--- a/examples/aws_efa/README.md
+++ b/examples/aws_efa/README.md
@@ -41,7 +41,7 @@ hyperpod-i-0da69b9076c7ff6a4   ml.p4d.24xlarge   8     4
 
 ## Access HyperPod and run distributed job with SkyPilot
 
-To accesss HyperPod and run distributed job with SkyPilot, see the SkyPilot [HyperPod example](../hyperpod-eks/).
+To access HyperPod and run distributed job with SkyPilot, see the SkyPilot [HyperPod example](https://github.com/skypilot-org/skypilot/blob/master/examples/hyperpod-eks).
 
 ### Adding EFA configurations in SkyPilot YAML
 
@@ -101,11 +101,11 @@ Check the following table for the GPU and EFA count mapping for AWS instance typ
 | g6e.48xlarge  | L40S:8   | 4    |
 
 
-Update the EFA number in the [nccl_efa.yaml](nccl_efa.yaml) for the GPUs you used.
+Update the EFA number in the [`nccl_efa.yaml`](https://github.com/skypilot-org/skypilot/blob/master/examples/aws_efa/nccl_efa.yaml) for the GPUs you use.
 
 ## Running NCCL test with EFA using SkyPilot
 
-Check the [nccl_efa.yaml](nccl_efa.yaml) for the complete SkyPilot cluster yaml cofnigurations.
+Check the [`nccl_efa.yaml`](https://github.com/skypilot-org/skypilot/blob/master/examples/aws_efa/nccl_efa.yaml) for the complete SkyPilot cluster yaml configurations.
 
 The `image_id` provides the environment setup for [NCCL](https://developer.nvidia.com/nccl) (NVIDIA Collective Communications Library) and EFA (Elastic Fabric Adapter).
 
@@ -121,7 +121,7 @@ SkyPilot will:
 3. Execute the NCCL performance test with EFA networking
 4. Output performance metrics showing the benefits of EFA for distributed training
 
-> [!NOTE]
+> **NOTE:**
 > We can turn off EFA with `nccl_efa.yaml` by passing an env:
 > ```bash
 > sky launch -c efa --env USE_EFA=false nccl_efa.yaml
diff --git a/examples/gcp_gpu_direct_tcpx/README.md b/examples/gcp_gpu_direct_tcpx/README.md
index 2cba5f8c5..1c51ec171 100644
--- a/examples/gcp_gpu_direct_tcpx/README.md
+++ b/examples/gcp_gpu_direct_tcpx/README.md
@@ -32,7 +32,7 @@ With `enable_gpu_direct: true`, SkyPilot automatically:
 
 ## Running NCCL Tests with GPUDirect-TCPX
 
-The complete configuration is available in `gpu_direct_tcpx.yaml`. The configuration includes:
+The complete configuration is available in [`gpu_direct_tcpx.yaml`](https://github.com/skypilot-org/skypilot/blob/master/examples/gcp_gpu_direct_tcpx/gpu_direct_tcpx.yaml). The configuration includes:
 - `image_id`: Pre-configured environment for NCCL testing
 - `instance_type`: Set to `a3-highgpu-8g`
 
@@ -43,7 +43,7 @@ sky launch -c tcpx gpu_direct_tcpx.yaml
 ```
 
 SkyPilot will:
-1. Deploy a GCP cluster with GPUDirect-TCPX-enabled nodes
+1. Deploy a GCP cluster with GPUDirect-TCPX enabled nodes
 2. Execute NCCL performance tests
 3. Output detailed performance metrics
 
diff --git a/sky/backends/backend.py b/sky/backends/backend.py
index ca40cb763..ca01c64af 100644
--- a/sky/backends/backend.py
+++ b/sky/backends/backend.py
@@ -37,8 +37,9 @@ class Backend(Generic[_ResourceHandleType]):
     ResourceHandle = ResourceHandle  # pylint: disable=invalid-name
 
     # --- APIs ---
-    def check_resources_fit_cluster(self, handle: _ResourceHandleType,
-                                    task: 'task_lib.Task') -> None:
+    def check_resources_fit_cluster(
+            self, handle: _ResourceHandleType,
+            task: 'task_lib.Task') -> Optional['resources.Resources']:
         """Check whether resources of the task are satisfied by cluster."""
         raise NotImplementedError
 
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 0463fc274..3a1333e30 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -606,7 +606,7 @@ def write_cluster_config(
     # other cases, we exclude the cloud from credential file uploads after
     # running required checks.
     assert cluster_name is not None
-    excluded_clouds = set()
+    excluded_clouds: Set[clouds.Cloud] = set()
     remote_identity_config = skypilot_config.get_nested(
         (str(cloud).lower(), 'remote_identity'), None)
     remote_identity = schemas.get_default_remote_identity(str(cloud).lower())
@@ -1557,7 +1557,8 @@ def check_owner_identity(cluster_name: str) -> None:
     if not isinstance(handle, backends.CloudVmRayResourceHandle):
         return
 
-    cloud = handle.launched_resources.cloud
+    launched_resources = handle.launched_resources.assert_launchable()
+    cloud = launched_resources.cloud
     user_identities = cloud.get_user_identities()
     owner_identity = record['owner']
     if user_identities is None:
@@ -1721,12 +1722,12 @@ def check_can_clone_disk_and_override_task(
                     'a new target cluster name.')
 
     new_task_resources = []
-    original_cloud = handle.launched_resources.cloud
+    launched_resources = handle.launched_resources.assert_launchable()
+    original_cloud = launched_resources.cloud
     original_cloud.check_features_are_supported(
-        handle.launched_resources,
+        launched_resources,
         {clouds.CloudImplementationFeatures.CLONE_DISK_FROM_CLUSTER})
 
-    assert original_cloud is not None, handle.launched_resources
     has_override = False
     has_disk_size_met = False
     has_cloud_met = False
@@ -1740,7 +1741,7 @@ def check_can_clone_disk_and_override_task(
             continue
         has_cloud_met = True
 
-        override_param = {}
+        override_param: Dict[str, Any] = {}
         if task_resources.cloud is None:
             override_param['cloud'] = original_cloud
         if task_resources.region is None:
@@ -1934,8 +1935,8 @@ def _update_cluster_status(cluster_name: str) -> Optional[Dict[str, Any]]:
         return global_user_state.get_cluster_from_name(cluster_name)
 
     # All cases below are transitioning the cluster to non-UP states.
-
-    if (not node_statuses and handle.launched_resources.cloud.STATUS_VERSION >=
+    launched_resources = handle.launched_resources.assert_launchable()
+    if (not node_statuses and launched_resources.cloud.STATUS_VERSION >=
             clouds.StatusVersion.SKYPILOT):
         # Note: launched_at is set during sky launch, even on an existing
         # cluster. This will catch the case where the cluster was terminated on
@@ -2470,7 +2471,7 @@ def is_controller_accessible(
           need_connection_check):
         # Check ssh connection if (1) controller is in INIT state, or (2) we failed to fetch the
         # status, both of which can happen when controller's status lock is held by another `sky jobs launch` or
-        # `sky serve up`. If we have controller's head_ip available and it is ssh-reachable,
+        # `sky serve up`. If we have controller's head_ip available and it is ssh-reachable,
         # we can allow access to the controller.
         ssh_credentials = ssh_credential_from_yaml(handle.cluster_yaml,
                                                    handle.docker_user,
@@ -2968,7 +2969,7 @@ def get_endpoints(cluster: str,
                              f'for cluster {cluster!r} with backend '
                              f'{get_backend_from_handle(handle).NAME}.')
 
-    launched_resources = handle.launched_resources
+    launched_resources = handle.launched_resources.assert_launchable()
     cloud = launched_resources.cloud
     try:
         cloud.check_features_are_supported(
@@ -2985,11 +2986,11 @@ def get_endpoints(cluster: str,
                                              head_ip=handle.head_ip,
                                              provider_config=config['provider'])
 
+    launched_resources = handle.launched_resources.assert_launchable()
     # Validation before returning the endpoints
     if port is not None:
         # If the requested endpoint was not to be exposed
-        port_set = resources_utils.port_ranges_to_set(
-            handle.launched_resources.ports)
+        port_set = resources_utils.port_ranges_to_set(launched_resources.ports)
         if port not in port_set:
             logger.warning(f'Port {port} is not exposed on '
                            f'cluster {cluster!r}.')
@@ -2998,8 +2999,7 @@ def get_endpoints(cluster: str,
         if port not in port_details:
             error_msg = (f'Port {port} not exposed yet. '
                          f'{_ENDPOINTS_RETRY_MESSAGE} ')
-            if handle.launched_resources.cloud.is_same_cloud(
-                    clouds.Kubernetes()):
+            if launched_resources.cloud.is_same_cloud(clouds.Kubernetes()):
                 # Add Kubernetes specific debugging info
                 error_msg += (kubernetes_utils.get_endpoint_debug_message())
             logger.warning(error_msg)
@@ -3008,7 +3008,7 @@ def get_endpoints(cluster: str,
     else:
         if not port_details:
             # If cluster had no ports to be exposed
-            if handle.launched_resources.ports is None:
+            if launched_resources.ports is None:
                 logger.warning(f'Cluster {cluster!r} does not have any '
                                'ports to be exposed.')
                 return {}
@@ -3017,8 +3017,7 @@ def get_endpoints(cluster: str,
             else:
                 error_msg = (f'No endpoints exposed yet. '
                              f'{_ENDPOINTS_RETRY_MESSAGE} ')
-                if handle.launched_resources.cloud.is_same_cloud(
-                        clouds.Kubernetes()):
+                if launched_resources.cloud.is_same_cloud(clouds.Kubernetes()):
                     # Add Kubernetes specific debugging info
                     error_msg += \
                         kubernetes_utils.get_endpoint_debug_message()
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index e1564e933..8ce9691c4 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -303,8 +303,6 @@ class RayCodeGen:
             from sky.skylet import autostop_lib
             from sky.skylet import constants
             from sky.skylet import job_lib
-            from sky.utils import context
-            from sky.utils import context_utils
             from sky.utils import log_utils
             from sky.utils import subprocess_utils
 
@@ -367,6 +365,7 @@ class RayCodeGen:
             # by ray.remote. This should be removed once we have a better way to
             # specify dependencies for ray.
             inspect.getsource(log_lib._ProcessingArgs),  # pylint: disable=protected-access
+            inspect.getsource(log_lib._get_context),  # pylint: disable=protected-access
             inspect.getsource(log_lib._handle_io_stream),  # pylint: disable=protected-access
             inspect.getsource(log_lib.process_subprocess_stream),
             inspect.getsource(log_lib.run_with_log),
@@ -1359,6 +1358,8 @@ class RetryingVmProvisioner(object):
         # Get previous cluster status
         cluster_exists = prev_cluster_status is not None
 
+        to_provision = to_provision.assert_launchable()
+
         assert to_provision.region is not None, (
             to_provision, 'region should have been set by the optimizer.')
         region = clouds.Region(to_provision.region)
@@ -2034,6 +2035,7 @@ class RetryingVmProvisioner(object):
                             f' that never expire or a service account.\033[0m')
                 logger.warning(warnings)
 
+        to_provision = to_provision.assert_launchable()
         # Retrying launchable resources.
         while True:
             try:
@@ -2163,9 +2165,10 @@ class RetryingVmProvisioner(object):
                 raise exceptions.ResourcesUnavailableError(
                     _RESOURCES_UNAVAILABLE_LOG + '\n' + table.get_string(),
                     failover_history=failover_history)
-            to_provision = task.best_resources
+            best_resources = task.best_resources
             assert task in self._dag.tasks, 'Internal logic error.'
-            assert to_provision is not None, task
+            assert best_resources is not None, task
+            to_provision = best_resources
         return config_dict
 
 
@@ -2431,19 +2434,21 @@ class CloudVmRayResourceHandle(backends.backend.ResourceHandle):
             self.cluster_yaml, self.docker_user, self.ssh_user)
         if avoid_ssh_control:
             ssh_credentials.pop('ssh_control_name', None)
+
+        launched_resources = self.launched_resources.assert_launchable()
         updated_to_skypilot_provisioner_after_provisioned = (
-            self.launched_resources.cloud.PROVISIONER_VERSION >=
+            launched_resources.cloud.PROVISIONER_VERSION >=
             clouds.ProvisionerVersion.SKYPILOT and
             self.cached_external_ips is not None and
             self.cached_cluster_info is None)
         if updated_to_skypilot_provisioner_after_provisioned:
             logger.debug(
-                f'{self.launched_resources.cloud} has been updated to the new '
+                f'{launched_resources.cloud} has been updated to the new '
                 f'provisioner after cluster {self.cluster_name} was '
                 f'provisioned. Cached IPs are used for connecting to the '
                 'cluster.')
         if (clouds.ProvisionerVersion.RAY_PROVISIONER_SKYPILOT_TERMINATOR >=
-                self.launched_resources.cloud.PROVISIONER_VERSION or
+                launched_resources.cloud.PROVISIONER_VERSION or
                 updated_to_skypilot_provisioner_after_provisioned):
             ip_list = (self.cached_external_ips
                        if force_cached else self.external_ips())
@@ -3147,8 +3152,8 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
             resources_utils.port_ranges_to_set(current_ports) -
             resources_utils.port_ranges_to_set(prev_ports))
         if open_new_ports:
-            cloud = handle.launched_resources.cloud
-            if not (cloud.OPEN_PORTS_VERSION <=
+            launched_resources = handle.launched_resources.assert_launchable()
+            if not (launched_resources.cloud.OPEN_PORTS_VERSION <=
                     clouds.OpenPortsVersion.LAUNCH_ONLY):
                 with rich_utils.safe_status(
                         ux_utils.spinner_message(
@@ -3252,9 +3257,10 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
         TODO: Delete COPY storage_mounts in task.sync_storage_mounts(), and
         assert here that all storage_mounts are MOUNT mode.
         """
+        launched_resources = handle.launched_resources.assert_launchable()
         with rich_utils.safe_status(ux_utils.spinner_message('Syncing files')):
             controller_utils.replace_skypilot_config_path_in_file_mounts(
-                handle.launched_resources.cloud, all_file_mounts)
+                launched_resources.cloud, all_file_mounts)
             self._execute_file_mounts(handle, all_file_mounts)
             self._execute_storage_mounts(handle, storage_mounts)
             self._set_storage_mounts_metadata(handle.cluster_name,
@@ -4154,7 +4160,8 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
         log_path = os.path.join(os.path.expanduser(self.log_dir),
                                 'teardown.log')
         log_abs_path = os.path.abspath(log_path)
-        cloud = handle.launched_resources.cloud
+        launched_resources = handle.launched_resources.assert_launchable()
+        cloud = launched_resources.cloud
         config = common_utils.read_yaml(handle.cluster_yaml)
         cluster_name = handle.cluster_name
         cluster_name_on_cloud = handle.cluster_name_on_cloud
@@ -4366,10 +4373,12 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
             # the right resource to provision the cluster.
             if handle.cluster_yaml is not None:
                 try:
-                    cloud = handle.launched_resources.cloud
+                    launched_resources = (
+                        handle.launched_resources.assert_launchable())
+                    cloud = launched_resources.cloud
                     config = common_utils.read_yaml(handle.cluster_yaml)
                     cloud.check_features_are_supported(
-                        handle.launched_resources,
+                        launched_resources,
                         {clouds.CloudImplementationFeatures.OPEN_PORTS})
                     provision_lib.cleanup_ports(repr(cloud),
                                                 cluster_name_on_cloud,
@@ -4711,6 +4720,8 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
             all_ports = resources_utils.port_set_to_ranges(current_ports_set |
                                                            requested_ports_set)
             to_provision = handle.launched_resources
+            assert to_provision is not None
+            to_provision = to_provision.assert_launchable()
             if (to_provision.cloud.OPEN_PORTS_VERSION <=
                     clouds.OpenPortsVersion.LAUNCH_ONLY):
                 if not requested_ports_set <= current_ports_set:
diff --git a/sky/clouds/aws.py b/sky/clouds/aws.py
index bb756d84c..22df098bf 100644
--- a/sky/clouds/aws.py
+++ b/sky/clouds/aws.py
@@ -440,18 +440,19 @@ class AWS(clouds.Cloud):
         region_name = region.name
         zone_names = [zone.name for zone in zones]
 
-        r = resources
-        # r.accelerators is cleared but .instance_type encodes the info.
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        # resources.accelerators is cleared but .instance_type encodes the info.
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
-        if r.extract_docker_image() is not None:
+        if resources.extract_docker_image() is not None:
             image_id_to_use = None
         else:
-            image_id_to_use = r.image_id
+            image_id_to_use = resources.image_id
         image_id = self._get_image_id(image_id_to_use, region_name,
-                                      r.instance_type)
+                                      resources.instance_type)
 
         disk_encrypted = skypilot_config.get_nested(('aws', 'disk_encrypted'),
                                                     False)
@@ -483,17 +484,17 @@ class AWS(clouds.Cloud):
                     'in `~/.sky/config.yaml`.')
 
         return {
-            'instance_type': r.instance_type,
+            'instance_type': resources.instance_type,
             'custom_resources': custom_resources,
             'disk_encrypted': disk_encrypted,
-            'use_spot': r.use_spot,
+            'use_spot': resources.use_spot,
             'region': region_name,
             'zones': ','.join(zone_names),
             'image_id': image_id,
             'security_group': security_group,
             'security_group_managed_by_skypilot':
                 str(security_group != user_security_group).lower(),
-            **AWS._get_disk_specs(r.disk_tier)
+            **AWS._get_disk_specs(resources.disk_tier)
         }
 
     def _get_feasible_launchable_resources(
@@ -971,6 +972,7 @@ class AWS(clouds.Cloud):
             botocore.exceptions.ClientError: error in Boto3 client request.
         """
 
+        resources = resources.assert_launchable()
         instance_type = resources.instance_type
         region = resources.region
         use_spot = resources.use_spot
diff --git a/sky/clouds/azure.py b/sky/clouds/azure.py
index 04b252b47..e3ac67295 100644
--- a/sky/clouds/azure.py
+++ b/sky/clouds/azure.py
@@ -325,9 +325,10 @@ class Azure(clouds.Cloud):
 
         region_name = region.name
 
-        r = resources
-        # r.accelerators is cleared but .instance_type encodes the info.
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        # resources.accelerators is cleared but .instance_type encodes the info.
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         acc_count = None
         if acc_dict is not None:
             acc_count = str(sum(acc_dict.values()))
@@ -339,8 +340,9 @@ class Azure(clouds.Cloud):
             # pylint: disable=import-outside-toplevel
             from sky.clouds.service_catalog import azure_catalog
             gen_version = azure_catalog.get_gen_version_from_instance_type(
-                r.instance_type)
-            image_id = self._get_default_image_tag(gen_version, r.instance_type)
+                resources.instance_type)
+            image_id = self._get_default_image_tag(gen_version,
+                                                   resources.instance_type)
         else:
             if None in resources.image_id:
                 image_id = resources.image_id[None]
@@ -407,18 +409,19 @@ class Azure(clouds.Cloud):
             """).split('\n')
 
         def _failover_disk_tier() -> Optional[resources_utils.DiskTier]:
-            if (r.disk_tier is not None and
-                    r.disk_tier != resources_utils.DiskTier.BEST):
-                return r.disk_tier
+            if (resources.disk_tier is not None and
+                    resources.disk_tier != resources_utils.DiskTier.BEST):
+                return resources.disk_tier
             # Failover disk tier from high to low. Default disk tier
             # (Premium_LRS, medium) only support s-series instance types,
             # so we failover to lower tiers for non-s-series.
             all_tiers = list(reversed(resources_utils.DiskTier))
             start_index = all_tiers.index(
-                Azure._translate_disk_tier(r.disk_tier))
+                Azure._translate_disk_tier(resources.disk_tier))
             while start_index < len(all_tiers):
                 disk_tier = all_tiers[start_index]
-                ok, _ = Azure.check_disk_tier(r.instance_type, disk_tier)
+                ok, _ = Azure.check_disk_tier(resources.instance_type,
+                                              disk_tier)
                 if ok:
                     return disk_tier
                 start_index += 1
@@ -426,11 +429,11 @@ class Azure(clouds.Cloud):
 
         disk_tier = _failover_disk_tier()
 
-        resources_vars = {
-            'instance_type': r.instance_type,
+        resources_vars: Dict[str, Any] = {
+            'instance_type': resources.instance_type,
             'custom_resources': custom_resources,
             'num_gpus': acc_count,
-            'use_spot': r.use_spot,
+            'use_spot': resources.use_spot,
             'region': region_name,
             # Azure does not support specific zones.
             'zones': None,
diff --git a/sky/clouds/cloud.py b/sky/clouds/cloud.py
index 0e391113c..2a17bf277 100644
--- a/sky/clouds/cloud.py
+++ b/sky/clouds/cloud.py
@@ -11,7 +11,8 @@ import collections
 import enum
 import math
 import typing
-from typing import Dict, Iterable, Iterator, List, Optional, Set, Tuple, Union
+from typing import (Any, Dict, Iterable, Iterator, List, Optional, Set, Tuple,
+                    Union)
 
 from typing_extensions import assert_never
 
@@ -302,7 +303,7 @@ class Cloud:
         zones: Optional[List['Zone']],
         num_nodes: int,
         dryrun: bool = False,
-    ) -> Dict[str, Optional[str]]:
+    ) -> Dict[str, Any]:
         """Converts planned sky.Resources to cloud-specific resource variables.
 
         These variables are used to fill the node type section (instance type,
@@ -721,7 +722,7 @@ class Cloud:
         Raises:
             ResourcesMismatchError: If the accelerator is not supported.
         """
-        assert resources.is_launchable(), resources
+        resources = resources.assert_launchable()
 
         def _equal_accelerators(
             acc_requested: Optional[Dict[str, Union[int, float]]],
diff --git a/sky/clouds/cudo.py b/sky/clouds/cudo.py
index 5e3d66fe0..fa71cf642 100644
--- a/sky/clouds/cudo.py
+++ b/sky/clouds/cudo.py
@@ -201,8 +201,9 @@ class Cudo(clouds.Cloud):
         dryrun: bool = False,
     ) -> Dict[str, Optional[str]]:
         del zones, cluster_name  # unused
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
diff --git a/sky/clouds/do.py b/sky/clouds/do.py
index c02f270e4..6c2a28fb3 100644
--- a/sky/clouds/do.py
+++ b/sky/clouds/do.py
@@ -181,8 +181,9 @@ class DO(clouds.Cloud):
             dryrun: bool = False) -> Dict[str, Optional[str]]:
         del zones, dryrun, cluster_name
 
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         if acc_dict is not None:
             custom_resources = json.dumps(acc_dict, separators=(',', ':'))
         else:
diff --git a/sky/clouds/fluidstack.py b/sky/clouds/fluidstack.py
index 1d2a9bad7..13666ec3c 100644
--- a/sky/clouds/fluidstack.py
+++ b/sky/clouds/fluidstack.py
@@ -188,9 +188,9 @@ class Fluidstack(clouds.Cloud):
     ) -> Dict[str, Optional[str]]:
 
         assert zones is None, 'FluidStack does not support zones.'
-
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
diff --git a/sky/clouds/gcp.py b/sky/clouds/gcp.py
index dcf4fdbcc..0d45f5431 100644
--- a/sky/clouds/gcp.py
+++ b/sky/clouds/gcp.py
@@ -1039,7 +1039,7 @@ class GCP(clouds.Cloud):
     @staticmethod
     def _check_instance_type_accelerators_combination(
             resources: 'resources.Resources') -> None:
-        assert resources.is_launchable(), resources
+        resources = resources.assert_launchable()
         service_catalog.check_accelerator_attachable_to_host(
             resources.instance_type, resources.accelerators, resources.zone,
             'gcp')
diff --git a/sky/clouds/ibm.py b/sky/clouds/ibm.py
index e46eceb24..256082730 100644
--- a/sky/clouds/ibm.py
+++ b/sky/clouds/ibm.py
@@ -175,7 +175,7 @@ class IBM(clouds.Cloud):
         zones: Optional[List['clouds.Zone']],
         num_nodes: int,
         dryrun: bool = False,
-    ) -> Dict[str, Optional[str]]:
+    ) -> Dict[str, Any]:
         """Converts planned sky.Resources to cloud-specific resource variables.
 
         These variables are used to fill the node type section (instance type,
@@ -204,30 +204,32 @@ class IBM(clouds.Cloud):
         # clouds implementing 'zones_provision_loop()'
         zone_names = [zone.name for zone in zones]  # type: ignore[union-attr]
 
-        r = resources
-        assert not r.use_spot, \
+        resources = resources.assert_launchable()
+        assert not resources.use_spot, \
             'IBM does not currently support spot instances in this framework'
 
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
-        instance_resources = _get_profile_resources(r.instance_type)
+        instance_resources = _get_profile_resources(resources.instance_type)
 
         worker_instance_type = get_cred_file_field('worker_instance_type',
-                                                   r.instance_type)
+                                                   resources.instance_type)
         worker_instance_resources = _get_profile_resources(worker_instance_type)
         # r.image_id: {clouds.Region:image_id} - property of Resources class
-        image_id = r.image_id[
-            region.name] if r.image_id else self.get_default_image(region_name)
+        image_id = resources.image_id[
+            region.name] if resources.image_id else self.get_default_image(
+                region_name)
 
         return {
-            'instance_type': r.instance_type,
+            'instance_type': resources.instance_type,
             'instance_resources': instance_resources,
             'worker_instance_type': worker_instance_type,
             'worker_instance_resources': worker_instance_resources,
             'custom_resources': custom_resources,
-            'use_spot': r.use_spot,
+            'use_spot': resources.use_spot,
             'region': region_name,
             'zones': ','.join(zone_names),
             'image_id': image_id,
diff --git a/sky/clouds/kubernetes.py b/sky/clouds/kubernetes.py
index 6a73f3318..af0f78f46 100644
--- a/sky/clouds/kubernetes.py
+++ b/sky/clouds/kubernetes.py
@@ -414,8 +414,9 @@ class Kubernetes(clouds.Cloud):
             context = region.name
         assert context is not None, 'No context found in kubeconfig'
 
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
diff --git a/sky/clouds/lambda_cloud.py b/sky/clouds/lambda_cloud.py
index 329cd472c..e6d8a745c 100644
--- a/sky/clouds/lambda_cloud.py
+++ b/sky/clouds/lambda_cloud.py
@@ -1,6 +1,6 @@
 """Lambda Cloud."""
 import typing
-from typing import Dict, Iterator, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
 
 from sky import clouds
 from sky.adaptors import common as adaptors_common
@@ -162,16 +162,16 @@ class Lambda(clouds.Cloud):
             region: 'clouds.Region',
             zones: Optional[List['clouds.Zone']],
             num_nodes: int,
-            dryrun: bool = False) -> Dict[str, Optional[str]]:
+            dryrun: bool = False) -> Dict[str, Any]:
         del cluster_name, dryrun  # Unused.
         assert zones is None, 'Lambda does not support zones.'
-
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
-        resources_vars = {
+        resources_vars: Dict[str, Any] = {
             'instance_type': resources.instance_type,
             'custom_resources': custom_resources,
             'region': region.name,
diff --git a/sky/clouds/nebius.py b/sky/clouds/nebius.py
index df44718b5..6f55f6682 100644
--- a/sky/clouds/nebius.py
+++ b/sky/clouds/nebius.py
@@ -1,7 +1,7 @@
 """ Nebius Cloud. """
 import os
 import typing
-from typing import Dict, Iterator, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
 
 from sky import clouds
 from sky.adaptors import nebius
@@ -191,12 +191,13 @@ class Nebius(clouds.Cloud):
             region: 'clouds.Region',
             zones: Optional[List['clouds.Zone']],
             num_nodes: int,
-            dryrun: bool = False) -> Dict[str, Optional[str]]:
+            dryrun: bool = False) -> Dict[str, Any]:
         del dryrun, cluster_name
         assert zones is None, ('Nebius does not support zones', zones)
 
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
         platform, _ = resources.instance_type.split('_')
@@ -209,7 +210,7 @@ class Nebius(clouds.Cloud):
             raise RuntimeError('Unsupported instance type for Nebius cloud:'
                                f' {resources.instance_type}')
 
-        resources_vars = {
+        resources_vars: Dict[str, Any] = {
             'instance_type': resources.instance_type,
             'custom_resources': custom_resources,
             'region': region.name,
diff --git a/sky/clouds/oci.py b/sky/clouds/oci.py
index 6a200f40a..3433aa119 100644
--- a/sky/clouds/oci.py
+++ b/sky/clouds/oci.py
@@ -23,7 +23,7 @@ History:
 import logging
 import os
 import typing
-from typing import Dict, Iterator, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
 
 from sky import clouds
 from sky import exceptions
@@ -213,10 +213,11 @@ class OCI(clouds.Cloud):
             region: Optional['clouds.Region'],
             zones: Optional[List['clouds.Zone']],
             num_nodes: int,
-            dryrun: bool = False) -> Dict[str, Optional[str]]:
+            dryrun: bool = False) -> Dict[str, Any]:
         del cluster_name, dryrun  # Unused.
         assert region is not None, resources
 
+        resources = resources.assert_launchable()
         acc_dict = self.get_accelerators_from_instance_type(
             resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
@@ -245,7 +246,8 @@ class OCI(clouds.Cloud):
             image_id, os_type = image_id.replace(' ', '').split(':')
 
         cpus = resources.cpus
-        instance_type_arr = resources.instance_type.split(
+        original_instance_type = resources.instance_type
+        instance_type_arr = original_instance_type.split(
             oci_utils.oci_config.INSTANCE_TYPE_RES_SPERATOR)
         instance_type = instance_type_arr[0]
 
@@ -259,12 +261,12 @@ class OCI(clouds.Cloud):
         else:
             if cpus is None:
                 cpus, mems = OCI.get_vcpus_mem_from_instance_type(
-                    resources.instance_type)
+                    original_instance_type)
                 resources = resources.copy(
                     cpus=cpus,
                     memory=mems,
                 )
-            if cpus is None and resources.instance_type.startswith(
+            if cpus is None and original_instance_type.startswith(
                     oci_utils.oci_config.VM_PREFIX):
                 cpus = f'{oci_utils.oci_config.DEFAULT_NUM_VCPUS}'
 
@@ -273,7 +275,7 @@ class OCI(clouds.Cloud):
             # If zone is not specified, try to get the first zone.
             if zones is None:
                 regions = service_catalog.get_region_zones_for_instance_type(
-                    instance_type=resources.instance_type,
+                    instance_type=original_instance_type,
                     use_spot=resources.use_spot,
                     clouds='oci')
                 zones = [r for r in iter(regions) if r.name == region.name
@@ -313,7 +315,7 @@ class OCI(clouds.Cloud):
             # OS type is not determined yet. So try to get it from vms.csv
             image_str = self._get_image_str(
                 image_id=resources.image_id,
-                instance_type=resources.instance_type,
+                instance_type=original_instance_type,
                 region=region.name)
 
             # pylint: disable=import-outside-toplevel
diff --git a/sky/clouds/paperspace.py b/sky/clouds/paperspace.py
index 4346798ab..c4c5667b0 100644
--- a/sky/clouds/paperspace.py
+++ b/sky/clouds/paperspace.py
@@ -185,8 +185,9 @@ class Paperspace(clouds.Cloud):
             dryrun: bool = False) -> Dict[str, Optional[str]]:
         del zones, dryrun, cluster_name
 
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
diff --git a/sky/clouds/runpod.py b/sky/clouds/runpod.py
index adbf9607a..2aad0a6ba 100644
--- a/sky/clouds/runpod.py
+++ b/sky/clouds/runpod.py
@@ -164,27 +164,27 @@ class RunPod(clouds.Cloud):
             region: 'clouds.Region',
             zones: Optional[List['clouds.Zone']],
             num_nodes: int,
-            dryrun: bool = False) -> Dict[str, Optional[str]]:
+            dryrun: bool = False) -> Dict[str, Optional[Union[str, bool]]]:
         del dryrun, cluster_name  # unused
         assert zones is not None, (region, zones)
 
         zone_names = [zone.name for zone in zones]
 
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
-        if r.image_id is None:
-            image_id = 'runpod/base:0.0.2'
-        elif r.extract_docker_image() is not None:
-            image_id = r.extract_docker_image()
+        if resources.image_id is None:
+            image_id: Optional[str] = 'runpod/base:0.0.2'
+        elif resources.extract_docker_image() is not None:
+            image_id = resources.extract_docker_image()
         else:
-            image_id = r.image_id[r.region]
+            image_id = resources.image_id[resources.region]
 
         instance_type = resources.instance_type
         use_spot = resources.use_spot
-
         hourly_cost = self.instance_type_to_hourly_cost(
             instance_type=instance_type, use_spot=use_spot)
 
diff --git a/sky/clouds/scp.py b/sky/clouds/scp.py
index 24fce5be4..4840d7007 100644
--- a/sky/clouds/scp.py
+++ b/sky/clouds/scp.py
@@ -189,12 +189,14 @@ class SCP(clouds.Cloud):
         del cluster_name, dryrun  # Unused.
         assert zones is None, 'SCP does not support zones.'
 
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
-        image_id = self._get_image_id(r.image_id, region.name, r.instance_type)
+        image_id = self._get_image_id(resources.image_id, region.name,
+                                      resources.instance_type)
         return {
             'instance_type': resources.instance_type,
             'custom_resources': custom_resources,
diff --git a/sky/clouds/vast.py b/sky/clouds/vast.py
index d2053cf60..5c29a8105 100644
--- a/sky/clouds/vast.py
+++ b/sky/clouds/vast.py
@@ -161,17 +161,18 @@ class Vast(clouds.Cloud):
             dryrun: bool = False) -> Dict[str, Optional[str]]:
         del zones, dryrun, cluster_name, num_nodes  # unused
 
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
-        if r.image_id is None:
-            image_id = 'vastai/base:0.0.2'
-        elif r.extract_docker_image() is not None:
-            image_id = r.extract_docker_image()
+        if resources.image_id is None:
+            image_id: Optional[str] = 'vastai/base:0.0.2'
+        elif resources.extract_docker_image() is not None:
+            image_id = resources.extract_docker_image()
         else:
-            image_id = r.image_id[r.region]
+            image_id = resources.image_id[resources.region]
 
         return {
             'instance_type': resources.instance_type,
diff --git a/sky/clouds/vsphere.py b/sky/clouds/vsphere.py
index fd261c22e..2a01a1f55 100644
--- a/sky/clouds/vsphere.py
+++ b/sky/clouds/vsphere.py
@@ -186,8 +186,10 @@ class Vsphere(clouds.Cloud):
         del cluster_name, dryrun  # unused
         assert zones is not None, (region, zones)
         zone_names = [zone.name for zone in zones]
-        r = resources
-        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+
+        resources = resources.assert_launchable()
+        acc_dict = self.get_accelerators_from_instance_type(
+            resources.instance_type)
         custom_resources = resources_utils.make_ray_custom_resources_str(
             acc_dict)
 
diff --git a/sky/core.py b/sky/core.py
index 2e5e4900c..356e66e5c 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -627,29 +627,26 @@ def autostop(
     )
     backend = backend_utils.get_backend_from_handle(handle)
 
+    resources = handle.launched_resources.assert_launchable()
     # Check cloud supports stopping spot instances
-    cloud = handle.launched_resources.cloud
-    assert cloud is not None, handle
+    cloud = resources.cloud
 
     if not isinstance(backend, backends.CloudVmRayBackend):
         raise exceptions.NotSupportedError(
             f'{operation} cluster {cluster_name!r} with backend '
             f'{backend.__class__.__name__!r} is not supported.')
-    cloud = handle.launched_resources.cloud
+
     # Check if autostop/autodown is required and supported
     if not is_cancel:
         try:
             if down:
                 cloud.check_features_are_supported(
-                    handle.launched_resources,
-                    {clouds.CloudImplementationFeatures.AUTODOWN})
+                    resources, {clouds.CloudImplementationFeatures.AUTODOWN})
             else:
                 cloud.check_features_are_supported(
-                    handle.launched_resources,
-                    {clouds.CloudImplementationFeatures.STOP})
+                    resources, {clouds.CloudImplementationFeatures.STOP})
                 cloud.check_features_are_supported(
-                    handle.launched_resources,
-                    {clouds.CloudImplementationFeatures.AUTOSTOP})
+                    resources, {clouds.CloudImplementationFeatures.AUTOSTOP})
         except exceptions.NotSupportedError as e:
             raise exceptions.NotSupportedError(
                 f'{colorama.Fore.YELLOW}{operation} on cluster '
diff --git a/sky/dashboard/src/pages/index.js b/sky/dashboard/src/pages/index.js
index b7dd5366a..311c11aee 100755
--- a/sky/dashboard/src/pages/index.js
+++ b/sky/dashboard/src/pages/index.js
@@ -5,7 +5,11 @@ export default function Index() {
   const router = useRouter();
 
   useEffect(() => {
-    router.push('/clusters');
+    if (router.asPath === '/') {
+      router.push('/clusters');
+    } else {
+      router.push(router.asPath);
+    }
   }, [router]);
 
   // Return null or a loading state while redirecting
diff --git a/sky/jobs/recovery_strategy.py b/sky/jobs/recovery_strategy.py
index ba2b4b283..a1333b7ba 100644
--- a/sky/jobs/recovery_strategy.py
+++ b/sky/jobs/recovery_strategy.py
@@ -86,8 +86,11 @@ class StrategyExecutor:
         # original task.resources
         task.set_resources(type(task.resources)(new_resources_list))
         if isinstance(job_recovery, dict):
-            job_recovery_name = job_recovery.pop(
+            name = job_recovery.pop(
                 'strategy', registry.JOBS_RECOVERY_STRATEGY_REGISTRY.default)
+            assert name is None or isinstance(name, str), (
+                name, 'The job recovery strategy name must be a string or None')
+            job_recovery_name: Optional[str] = name
             max_restarts_on_errors = job_recovery.pop('max_restarts_on_errors',
                                                       0)
         else:
diff --git a/sky/optimizer.py b/sky/optimizer.py
index 3c2f2c320..f4a9fa035 100644
--- a/sky/optimizer.py
+++ b/sky/optimizer.py
@@ -671,7 +671,7 @@ class Optimizer:
         plan: Dict[task_lib.Task, resources_lib.Resources],
     ) -> float:
         """Estimates the total cost of running the DAG by the plan."""
-        total_cost = 0
+        total_cost = 0.
         for node in topo_order:
             resources = plan[node]
             if node.time_estimator_func is None:
@@ -777,10 +777,13 @@ class Optimizer:
             accelerators = resources.get_accelerators_str()
             spot = resources.get_spot_str()
             cloud = resources.cloud
-            vcpus, mem = cloud.get_vcpus_mem_from_instance_type(
+            assert cloud is not None, 'Cloud must be specified'
+            assert (resources.instance_type is not None), \
+                'Instance type must be specified'
+            vcpus_, mem_ = cloud.get_vcpus_mem_from_instance_type(
                 resources.instance_type)
 
-            def format_number(x):
+            def format_number(x: Optional[float]) -> str:
                 if x is None:
                     return '-'
                 elif x.is_integer():
@@ -788,8 +791,8 @@ class Optimizer:
                 else:
                     return f'{x:.1f}'
 
-            vcpus = format_number(vcpus)
-            mem = format_number(mem)
+            vcpus = format_number(vcpus_)
+            mem = format_number(mem_)
 
             if resources.zone is None:
                 region_or_zone = resources.region
@@ -814,11 +817,12 @@ class Optimizer:
 
             accelerators = resources.get_accelerators_str()
             spot = resources.get_spot_str()
+            resources = resources.assert_launchable()
             cloud = resources.cloud
-            vcpus, mem = cloud.get_vcpus_mem_from_instance_type(
+            vcpus_, mem_ = cloud.get_vcpus_mem_from_instance_type(
                 resources.instance_type)
 
-            def format_number(x):
+            def format_number(x: Optional[float]) -> str:
                 if x is None:
                     return '-'
                 elif x.is_integer():
@@ -826,8 +830,8 @@ class Optimizer:
                 else:
                     return f'{x:.1f}'
 
-            vcpus = format_number(vcpus)
-            mem = format_number(mem)
+            vcpus = format_number(vcpus_)
+            mem = format_number(mem_)
 
             if resources.zone is None:
                 region_or_zone = resources.region
@@ -1195,10 +1199,12 @@ def _check_specified_clouds(dag: 'dag_lib.Dag') -> None:
             all_clouds_specified.add(cloud_str)
 
         # Explicitly check again to update the enabled cloud list.
-        sky_check.check_capability(sky_cloud.CloudCapability.COMPUTE,
-                                   quiet=True,
-                                   clouds=list(clouds_need_recheck -
-                                               global_disabled_clouds))
+        clouds_to_check_again = list(clouds_need_recheck -
+                                     global_disabled_clouds)
+        if len(clouds_to_check_again) > 0:
+            sky_check.check_capability(sky_cloud.CloudCapability.COMPUTE,
+                                       quiet=True,
+                                       clouds=clouds_to_check_again)
         enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
             capability=sky_cloud.CloudCapability.COMPUTE,
             raise_if_no_cloud_access=True)
diff --git a/sky/provision/kubernetes/utils.py b/sky/provision/kubernetes/utils.py
index b108d8c26..b0cd00b25 100644
--- a/sky/provision/kubernetes/utils.py
+++ b/sky/provision/kubernetes/utils.py
@@ -1736,9 +1736,16 @@ class KubernetesInstanceType:
     @staticmethod
     def is_valid_instance_type(name: str) -> bool:
         """Returns whether the given name is a valid instance type."""
+        # Before https://github.com/skypilot-org/skypilot/pull/4756,
+        # the accelerators are appended with format "--{a}{type}",
+        # e.g. "4CPU--16GB--1V100".
+        # Check both patterns to keep backward compatibility.
+        # TODO(romilb): Backward compatibility, remove after 0.11.0.
+        prev_pattern = re.compile(
+            r'^(\d+(\.\d+)?CPU--\d+(\.\d+)?GB)(--\d+\S+)?$')
         pattern = re.compile(
             r'^(\d+(\.\d+)?CPU--\d+(\.\d+)?GB)(--[\w\d-]+:\d+)?$')
-        return bool(pattern.match(name))
+        return bool(pattern.match(name)) or bool(prev_pattern.match(name))
 
     @classmethod
     def _parse_instance_type(
@@ -1755,6 +1762,11 @@ class KubernetesInstanceType:
             r'^(?P<cpus>\d+(\.\d+)?)CPU--(?P<memory>\d+(\.\d+)?)GB(?:--(?P<accelerator_type>[\w\d-]+):(?P<accelerator_count>\d+))?$'  # pylint: disable=line-too-long
         )
         match = pattern.match(name)
+        # TODO(romilb): Backward compatibility, remove after 0.11.0.
+        prev_pattern = re.compile(
+            r'^(?P<cpus>\d+(\.\d+)?)CPU--(?P<memory>\d+(\.\d+)?)GB(?:--(?P<accelerator_count>\d+)(?P<accelerator_type>\S+))?$'  # pylint: disable=line-too-long
+        )
+        prev_match = prev_pattern.match(name)
         if match:
             cpus = float(match.group('cpus'))
             memory = float(match.group('memory'))
@@ -1769,6 +1781,19 @@ class KubernetesInstanceType:
                 accelerator_count = None
                 accelerator_type = None
             return cpus, memory, accelerator_count, accelerator_type
+        # TODO(romilb): Backward compatibility, remove after 0.11.0.
+        elif prev_match:
+            cpus = float(prev_match.group('cpus'))
+            memory = float(prev_match.group('memory'))
+            accelerator_count = prev_match.group('accelerator_count')
+            accelerator_type = prev_match.group('accelerator_type')
+            if accelerator_count:
+                accelerator_count = int(accelerator_count)
+                accelerator_type = str(accelerator_type).replace('_', ' ')
+            else:
+                accelerator_count = None
+                accelerator_type = None
+            return cpus, memory, accelerator_count, accelerator_type
         else:
             raise ValueError(f'Invalid instance name: {name}')
 
diff --git a/sky/resources.py b/sky/resources.py
index ced4c6c29..6e9d72643 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -1,6 +1,7 @@
 """Resources: compute requirements of Tasks."""
 import dataclasses
 import textwrap
+import typing
 from typing import Any, Dict, List, Optional, Set, Tuple, Union
 
 import colorama
@@ -59,13 +60,14 @@ class Resources:
         instance_type: Optional[str] = None,
         cpus: Union[None, int, float, str] = None,
         memory: Union[None, int, float, str] = None,
-        accelerators: Union[None, str, Dict[str, int]] = None,
+        accelerators: Union[None, str, Dict[str, Union[int, float]]] = None,
         accelerator_args: Optional[Dict[str, str]] = None,
         use_spot: Optional[bool] = None,
-        job_recovery: Optional[Union[Dict[str, Union[str, int]], str]] = None,
+        job_recovery: Optional[Union[Dict[str, Optional[Union[str, int]]],
+                                     str]] = None,
         region: Optional[str] = None,
         zone: Optional[str] = None,
-        image_id: Union[Dict[str, str], str, None] = None,
+        image_id: Union[Dict[Optional[str], str], str, None] = None,
         disk_size: Optional[int] = None,
         disk_tier: Optional[Union[str, resources_utils.DiskTier]] = None,
         ports: Optional[Union[int, str, List[str], Tuple[str]]] = None,
@@ -177,7 +179,8 @@ class Resources:
 
         self._use_spot_specified = use_spot is not None
         self._use_spot = use_spot if use_spot is not None else False
-        self._job_recovery: Optional[Dict[str, Union[str, int]]] = None
+        self._job_recovery: Optional[Dict[str, Optional[Union[str,
+                                                              int]]]] = None
         if job_recovery is not None:
             if isinstance(job_recovery, str):
                 job_recovery = {'strategy': job_recovery}
@@ -188,7 +191,7 @@ class Resources:
             if strategy_name == 'none':
                 self._job_recovery = None
             else:
-                if strategy_name is not None:
+                if isinstance(strategy_name, str):
                     job_recovery['strategy'] = strategy_name.upper()
                 self._job_recovery = job_recovery
 
@@ -201,7 +204,7 @@ class Resources:
         else:
             self._disk_size = _DEFAULT_DISK_SIZE_GB
 
-        self._image_id = image_id
+        self._image_id: Optional[Dict[Optional[str], str]] = None
         if isinstance(image_id, str):
             self._image_id = {self._region: image_id.strip()}
         elif isinstance(image_id, dict):
@@ -209,8 +212,11 @@ class Resources:
                 self._image_id = {self._region: image_id[None].strip()}
             else:
                 self._image_id = {
-                    k.strip(): v.strip() for k, v in image_id.items()
+                    typing.cast(str, k).strip(): v.strip()
+                    for k, v in image_id.items()
                 }
+        else:
+            self._image_id = image_id
         self._is_image_managed = _is_image_managed
 
         if isinstance(disk_tier, str):
@@ -228,7 +234,7 @@ class Resources:
             if isinstance(ports, tuple):
                 ports = list(ports)
             if not isinstance(ports, list):
-                ports = [ports]
+                ports = [str(ports)]
             ports = resources_utils.simplify_ports(
                 [str(port) for port in ports])
             if not ports:
@@ -250,7 +256,7 @@ class Resources:
         self._requires_fuse = _requires_fuse
 
         self._cluster_config_overrides = _cluster_config_overrides
-        self._cached_repr = None
+        self._cached_repr: Optional[str] = None
 
         self._set_cpus(cpus)
         self._set_memory(memory)
@@ -378,19 +384,19 @@ class Resources:
         return repr_str
 
     @property
-    def cloud(self):
+    def cloud(self) -> Optional[clouds.Cloud]:
         return self._cloud
 
     @property
-    def region(self):
+    def region(self) -> Optional[str]:
         return self._region
 
     @property
-    def zone(self):
+    def zone(self) -> Optional[str]:
         return self._zone
 
     @property
-    def instance_type(self):
+    def instance_type(self) -> Optional[str]:
         return self._instance_type
 
     @property
@@ -444,7 +450,7 @@ class Resources:
         return None
 
     @property
-    def accelerator_args(self) -> Optional[Dict[str, str]]:
+    def accelerator_args(self) -> Optional[Dict[str, Any]]:
         return self._accelerator_args
 
     @property
@@ -456,7 +462,7 @@ class Resources:
         return self._use_spot_specified
 
     @property
-    def job_recovery(self) -> Optional[Dict[str, Union[str, int]]]:
+    def job_recovery(self) -> Optional[Dict[str, Optional[Union[str, int]]]]:
         return self._job_recovery
 
     @property
@@ -464,11 +470,11 @@ class Resources:
         return self._disk_size
 
     @property
-    def image_id(self) -> Optional[Dict[str, str]]:
+    def image_id(self) -> Optional[Dict[Optional[str], str]]:
         return self._image_id
 
     @property
-    def disk_tier(self) -> resources_utils.DiskTier:
+    def disk_tier(self) -> Optional[resources_utils.DiskTier]:
         return self._disk_tier
 
     @property
@@ -489,16 +495,22 @@ class Resources:
             return False
         return self._requires_fuse
 
+    def set_requires_fuse(self, value: bool) -> None:
+        """Sets whether this resource requires FUSE mounting support.
+
+        Args:
+            value: Whether the resource requires FUSE mounting support.
+        """
+        # TODO(zeping): This violates the immutability of Resources.
+        #  Refactor to use Resources.copy instead.
+        self._requires_fuse = value
+
     @property
     def cluster_config_overrides(self) -> Dict[str, Any]:
         if self._cluster_config_overrides is None:
             return {}
         return self._cluster_config_overrides
 
-    @requires_fuse.setter
-    def requires_fuse(self, value: Optional[bool]) -> None:
-        self._requires_fuse = value
-
     @property
     def docker_login_config(self) -> Optional[docker_utils.DockerLoginConfig]:
         return self._docker_login_config
@@ -572,8 +584,8 @@ class Resources:
 
     def _set_accelerators(
         self,
-        accelerators: Union[None, str, Dict[str, int]],
-        accelerator_args: Optional[Dict[str, str]],
+        accelerators: Union[None, str, Dict[str, Union[int, float]]],
+        accelerator_args: Optional[Dict[str, Any]],
     ) -> None:
         """Sets accelerators.
 
@@ -608,10 +620,11 @@ class Resources:
                         self._cloud = clouds.Kubernetes()
                     else:
                         self._cloud = clouds.GCP()
-                assert (self.cloud.is_same_cloud(clouds.GCP()) or
-                        self.cloud.is_same_cloud(clouds.Kubernetes())), (
-                            'Cloud must be GCP or Kubernetes for TPU '
-                            'accelerators.')
+                assert self.cloud is not None and (
+                    self.cloud.is_same_cloud(clouds.GCP()) or
+                    self.cloud.is_same_cloud(clouds.Kubernetes())), (
+                        'Cloud must be GCP or Kubernetes for TPU '
+                        'accelerators.')
 
                 if accelerator_args is None:
                     accelerator_args = {}
@@ -645,15 +658,28 @@ class Resources:
                                     'Cannot specify instance type (got '
                                     f'{self.instance_type!r}) for TPU VM.')
 
-        self._accelerators = accelerators
-        self._accelerator_args = accelerator_args
+        self._accelerators: Optional[Dict[str, Union[int,
+                                                     float]]] = accelerators
+        self._accelerator_args: Optional[Dict[str, Any]] = accelerator_args
 
     def is_launchable(self) -> bool:
+        """Returns whether the resource is launchable."""
         return self.cloud is not None and self._instance_type is not None
 
+    def assert_launchable(self) -> 'LaunchableResources':
+        """A workaround to make mypy understand that is_launchable() is true.
+
+        Note: The `cast` to `LaunchableResources` is only for static type
+        checking with MyPy. At runtime, the Python interpreter does not enforce
+        types, and the returned object will still be an instance of `Resources`.
+        """
+        assert self.is_launchable(), self
+        return typing.cast(LaunchableResources, self)
+
     def need_cleanup_after_preemption_or_failure(self) -> bool:
         """Whether a resource needs cleanup after preemption or failure."""
         assert self.is_launchable(), self
+        assert self.cloud is not None, 'Cloud must be specified'
         return self.cloud.need_cleanup_after_preemption_or_failure(self)
 
     def _try_canonicalize_accelerators(self) -> None:
@@ -710,10 +736,10 @@ class Resources:
                     else:
                         table = log_utils.create_table(['Cloud', 'Hint'])
                         table.add_row(['-----', '----'])
-                        for cloud, error in cloud_to_errors.items():
+                        for cloud_msg, error in cloud_to_errors.items():
                             reason_str = '\n'.join(textwrap.wrap(
                                 str(error), 80))
-                            table.add_row([str(cloud), reason_str])
+                            table.add_row([cloud_msg, reason_str])
                         hint = table.get_string()
                     raise ValueError(
                         f'Invalid (region {self._region!r}, zone '
@@ -745,11 +771,13 @@ class Resources:
         ssh_proxy_command dict with region names as keys).
         """
         assert self.is_launchable(), self
-
-        regions = self._cloud.regions_with_offering(self._instance_type,
-                                                    self.accelerators,
-                                                    self._use_spot,
-                                                    self._region, self._zone)
+        assert self.cloud is not None, 'Cloud must be specified'
+        assert self._instance_type is not None, (
+            'Instance type must be specified')
+        regions = self.cloud.regions_with_offering(self._instance_type,
+                                                   self.accelerators,
+                                                   self._use_spot, self._region,
+                                                   self._zone)
         if self._image_id is not None and None not in self._image_id:
             regions = [r for r in regions if r.name in self._image_id]
 
@@ -849,6 +877,10 @@ class Resources:
             cpus, mem = self.cloud.get_vcpus_mem_from_instance_type(
                 self._instance_type)
             if self._cpus is not None:
+                assert cpus is not None, (
+                    f'Can\'t get vCPUs from instance type: '
+                    f'{self._instance_type}, check catalog or '
+                    f'specify cpus directly.')
                 if self._cpus.endswith('+'):
                     if cpus < float(self._cpus[:-1]):
                         with ux_utils.print_exception_no_traceback():
@@ -863,6 +895,10 @@ class Resources:
                             f'number of vCPUs. {self.instance_type} has {cpus} '
                             f'vCPUs, but {self._cpus} is requested.')
             if self.memory is not None:
+                assert mem is not None, (
+                    f'Can\'t get memory from instance type: '
+                    f'{self._instance_type}, check catalog or '
+                    f'specify memory directly.')
                 if self.memory.endswith(('+', 'x')):
                     if mem < float(self.memory[:-1]):
                         with ux_utils.print_exception_no_traceback():
@@ -886,6 +922,8 @@ class Resources:
         if self._job_recovery is None or self._job_recovery['strategy'] is None:
             return
         # Validate the job recovery strategy
+        assert isinstance(self._job_recovery['strategy'],
+                          str), 'Job recovery strategy must be a string'
         registry.JOBS_RECOVERY_STRATEGY_REGISTRY.from_str(
             self._job_recovery['strategy'])
 
@@ -920,7 +958,7 @@ class Resources:
                     'Cloud must be specified when image_id is provided.')
 
         try:
-            self._cloud.check_features_are_supported(
+            self.cloud.check_features_are_supported(
                 self,
                 requested_features={
                     clouds.CloudImplementationFeatures.IMAGE_ID
@@ -943,14 +981,14 @@ class Resources:
         # Check the image_id's are valid.
         for region, image_id in self._image_id.items():
             if (image_id.startswith('skypilot:') and
-                    not self._cloud.is_image_tag_valid(image_id, region)):
+                    not self.cloud.is_image_tag_valid(image_id, region)):
                 region_str = f' ({region})' if region else ''
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         f'Image tag {image_id!r} is not valid, please make sure'
                         f' the tag exists in {self._cloud}{region_str}.')
 
-            if (self._cloud.is_same_cloud(clouds.AWS()) and
+            if (self.cloud.is_same_cloud(clouds.AWS()) and
                     not image_id.startswith('skypilot:') and region is None):
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
@@ -1055,6 +1093,9 @@ class Resources:
         """Returns cost in USD for the runtime in seconds."""
         hours = seconds / 3600
         # Instance.
+        assert self.cloud is not None, 'Cloud must be specified'
+        assert self._instance_type is not None, (
+            'Instance type must be specified')
         hourly_cost = self.cloud.instance_type_to_hourly_cost(
             self._instance_type, self.use_spot, self._region, self._zone)
         # Accelerators (if any).
@@ -1099,6 +1140,7 @@ class Resources:
         docker_image = self.extract_docker_image()
 
         # Cloud specific variables
+        assert self.cloud is not None, 'Cloud must be specified'
         cloud_specific_variables = self.cloud.make_deploy_resources_variables(
             self, cluster_name, region, zones, num_nodes, dryrun)
 
@@ -1153,9 +1195,12 @@ class Resources:
         specific_reservations = set(
             skypilot_config.get_nested(
                 (str(self.cloud).lower(), 'specific_reservations'), set()))
+
+        assert (self.cloud is not None and self.instance_type is not None and
+                self.region
+                is not None), ('Cloud, instance type, region must be specified')
         return self.cloud.get_reservations_available_resources(
-            self._instance_type, self._region, self._zone,
-            specific_reservations)
+            self.instance_type, self.region, self.zone, specific_reservations)
 
     def less_demanding_than(
         self,
@@ -1175,6 +1220,9 @@ class Resources:
         if isinstance(other, list):
             resources_list = [self.less_demanding_than(o) for o in other]
             return requested_num_nodes <= sum(resources_list)
+
+        assert other.cloud is not None, 'Other cloud must be specified'
+
         if self.cloud is not None and not self.cloud.is_same_cloud(other.cloud):
             return False
         # self.cloud <= other.cloud
@@ -1263,6 +1311,7 @@ class Resources:
         If a field in `blocked` is None, it should be considered as a wildcard
         for that field.
         """
+        assert self.cloud is not None, 'Cloud must be specified'
         is_matched = True
         if (blocked.cloud is not None and
                 not self.cloud.is_same_cloud(blocked.cloud)):
@@ -1301,7 +1350,7 @@ class Resources:
         use_spot = self.use_spot if self._use_spot_specified else None
 
         current_override_configs = self._cluster_config_overrides
-        if self._cluster_config_overrides is None:
+        if current_override_configs is None:
             current_override_configs = {}
         new_override_configs = override.pop('_cluster_config_overrides', {})
         overlaid_configs = skypilot_config.overlay_skypilot_config(
@@ -1734,3 +1783,38 @@ class Resources:
                 '_docker_username_for_runpod', None)
 
         self.__dict__.update(state)
+
+
+class LaunchableResources(Resources):
+    """A class representing resources that can be launched on a cloud provider.
+
+    This class is primarily a type hint for MyPy to indicate that an instance
+    of `Resources` is launchable (i.e., `cloud` and `instance_type` are not
+    None). It should not be instantiated directly.
+    """
+
+    def __init__(self, *args, **kwargs) -> None:  # pylint: disable=super-init-not-called,unused-argument
+        assert False, (
+            'LaunchableResources should not be instantiated directly. '
+            'It is only used for type checking by MyPy.')
+
+    @property
+    def cloud(self) -> clouds.Cloud:
+        assert self._cloud is not None, 'Cloud must be specified'
+        return self._cloud
+
+    @property
+    def instance_type(self) -> str:
+        assert self._instance_type is not None, (
+            'Instance type must be specified')
+        return self._instance_type
+
+    def copy(self, **override) -> 'LaunchableResources':
+        """Ensure MyPy understands the return type is LaunchableResources.
+
+        This method is not expected to be called at runtime, as
+        LaunchableResources should not be directly instantiated. It primarily
+        serves as a type hint for static analysis.
+        """
+        self.assert_launchable()
+        return typing.cast(LaunchableResources, super().copy(**override))
diff --git a/sky/serve/spot_placer.py b/sky/serve/spot_placer.py
index 74bf0b5a5..79751b352 100644
--- a/sky/serve/spot_placer.py
+++ b/sky/serve/spot_placer.py
@@ -46,6 +46,8 @@ class Location:
 
     @classmethod
     def from_resources(cls, resources: 'resources_lib.Resources') -> 'Location':
+        assert resources.cloud is not None, 'Cloud must be specified'
+        assert resources.region is not None, 'Region must be specified'
         return cls(resources.cloud, resources.region, resources.zone)
 
     def to_dict(self) -> Dict[str, Any]:
@@ -147,6 +149,7 @@ def _get_possible_location_from_task(task: 'task_lib.Task') -> List[Location]:
                 cloud_str = str(launchable.cloud)
                 region = launchable.region
                 zone = launchable.zone
+                assert region is not None, 'Region must be specified'
                 if (cloud_str not in location_requirements and
                         location_requirements):
                     continue
diff --git a/sky/server/server.py b/sky/server/server.py
index 395a675bd..9668f5dd7 100644
--- a/sky/server/server.py
+++ b/sky/server/server.py
@@ -9,6 +9,7 @@ import logging
 import multiprocessing
 import os
 import pathlib
+import posixpath
 import re
 import shutil
 import sys
@@ -167,8 +168,36 @@ class InternalDashboardPrefixMiddleware(
         return await call_next(request)
 
 
+class CacheControlStaticMiddleware(starlette.middleware.base.BaseHTTPMiddleware
+                                  ):
+    """Middleware to add cache control headers to static files."""
+
+    async def dispatch(self, request: fastapi.Request, call_next):
+        if request.url.path.startswith('/dashboard/_next'):
+            response = await call_next(request)
+            response.headers['Cache-Control'] = 'max-age=3600'
+            return response
+        return await call_next(request)
+
+
+class PathCleanMiddleware(starlette.middleware.base.BaseHTTPMiddleware):
+    """Middleware to check the path of requests."""
+
+    async def dispatch(self, request: fastapi.Request, call_next):
+        if request.url.path.startswith('/dashboard/'):
+            # If the requested path is not relative to the expected directory,
+            # then the user is attempting path traversal, so deny the request.
+            parent = pathlib.Path('/dashboard')
+            request_path = pathlib.Path(posixpath.normpath(request.url.path))
+            if not _is_relative_to(request_path, parent):
+                raise fastapi.HTTPException(status_code=403, detail='Forbidden')
+        return await call_next(request)
+
+
 app = fastapi.FastAPI(prefix='/api/v1', debug=True, lifespan=lifespan)
 app.add_middleware(InternalDashboardPrefixMiddleware)
+app.add_middleware(PathCleanMiddleware)
+app.add_middleware(CacheControlStaticMiddleware)
 app.add_middleware(
     cors.CORSMiddleware,
     # TODO(zhwu): in production deployment, we should restrict the allowed
@@ -1130,25 +1159,28 @@ async def complete_storage_name(incomplete: str,) -> List[str]:
     return global_user_state.get_storage_names_start_with(incomplete)
 
 
-# Add a route to serve static files
-@app.get('/{full_path:path}')
-async def serve_static_or_dashboard(full_path: str):
-    """Serves static files for any unmatched routes.
+@app.get('/dashboard/{full_path:path}')
+async def serve_dashboard(full_path: str):
+    """Serves the Next.js dashboard application.
 
-    Handles the /dashboard prefix from Next.js configuration.
-    """
-    # Check if the path starts with 'dashboard/' and remove it if it does
-    if full_path.startswith('dashboard/'):
-        full_path = full_path[len('dashboard/'):]
+    Args:
+        full_path: The path requested by the client.
+        e.g. /clusters, /jobs
+
+    Returns:
+        FileResponse for static files or index.html for client-side routing.
 
-    # Try to serve the file directly from the out directory first
+    Raises:
+        HTTPException: If the path is invalid or file not found.
+    """
+    # Try to serve the staticfile directly e.g. /skypilot.svg,
+    # /favicon.ico, and /_next/, etc.
     file_path = os.path.join(server_constants.DASHBOARD_DIR, full_path)
     if os.path.isfile(file_path):
         return fastapi.responses.FileResponse(file_path)
 
-    # If file not found, serve the index.html for client-side routing.
-    # For example, the non-matched arbitrary route (/ or /test) from
-    # client will be redirected to the index.html.
+    # Serve index.html for client-side routing
+    # e.g. /clusters, /jobs
     index_path = os.path.join(server_constants.DASHBOARD_DIR, 'index.html')
     try:
         with open(index_path, 'r', encoding='utf-8') as f:
@@ -1159,6 +1191,12 @@ async def serve_static_or_dashboard(full_path: str):
         raise fastapi.HTTPException(status_code=500, detail=str(e))
 
 
+# Redirect the root path to dashboard
+@app.get('/')
+async def root():
+    return fastapi.responses.RedirectResponse(url='/dashboard/')
+
+
 if __name__ == '__main__':
     import uvicorn
 
diff --git a/sky/skylet/log_lib.py b/sky/skylet/log_lib.py
index d2a45edd4..878ee26a4 100644
--- a/sky/skylet/log_lib.py
+++ b/sky/skylet/log_lib.py
@@ -62,6 +62,16 @@ class _ProcessingArgs:
         self.streaming_prefix = streaming_prefix
 
 
+def _get_context():
+    # TODO(aylei): remove this after we drop the backward-compatibility for
+    # 0.9.x in 0.12.0
+    # Keep backward-compatibility for the old version of SkyPilot runtimes.
+    if 'context' in globals():
+        return context.get()
+    else:
+        return None
+
+
 def _handle_io_stream(io_stream, out_stream, args: _ProcessingArgs):
     """Process the stream of a process."""
     out_io = io.TextIOWrapper(io_stream,
@@ -80,7 +90,7 @@ def _handle_io_stream(io_stream, out_stream, args: _ProcessingArgs):
     with open(args.log_path, 'a', encoding='utf-8') as fout:
         with line_processor:
             while True:
-                ctx = context.get()
+                ctx = _get_context()
                 if ctx is not None and ctx.is_canceled():
                     return
                 line = out_io.readline()
@@ -139,7 +149,6 @@ def process_subprocess_stream(proc, stdout_stream_handler,
     return stdout, stderr
 
 
-@context_utils.cancellation_guard
 def run_with_log(
     cmd: Union[List[str], str],
     log_path: str,
@@ -181,7 +190,7 @@ def run_with_log(
     # Redirect stderr to stdout when using ray, to preserve the order of
     # stdout and stderr.
     stdout_arg = stderr_arg = None
-    ctx = context.get()
+    ctx = _get_context()
     if process_stream or ctx is not None:
         # Capture stdout/stderr of the subprocess if:
         # 1. Post-processing is needed (process_stream=True)
diff --git a/sky/skylet/log_lib.pyi b/sky/skylet/log_lib.pyi
index c7028e121..204867f98 100644
--- a/sky/skylet/log_lib.pyi
+++ b/sky/skylet/log_lib.pyi
@@ -11,6 +11,7 @@ from typing_extensions import Literal
 from sky import sky_logging as sky_logging
 from sky.skylet import constants as constants
 from sky.skylet import job_lib as job_lib
+from sky.utils import context
 from sky.utils import log_utils as log_utils
 
 SKY_LOG_WAITING_GAP_SECONDS: int = ...
@@ -41,6 +42,10 @@ class _ProcessingArgs:
         ...
 
 
+def _get_context() -> Optional[context.Context]:
+    ...
+
+
 def _handle_io_stream(io_stream, out_stream, args: _ProcessingArgs):
     ...
 
diff --git a/sky/task.py b/sky/task.py
index 14673c976..4b3beb8cd 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -165,7 +165,8 @@ def _with_docker_login_config(
                            f'ignored.{colorama.Style.RESET_ALL}')
             return resources
         # Already checked in extract_docker_image
-        assert len(resources.image_id) == 1, resources.image_id
+        assert resources.image_id is not None and len(
+            resources.image_id) == 1, resources.image_id
         region = list(resources.image_id.keys())[0]
         return resources.copy(image_id={region: 'docker:' + docker_image},
                               _docker_login_config=docker_login_config)
@@ -775,7 +776,7 @@ class Task:
         for _, storage_obj in self.storage_mounts.items():
             if storage_obj.mode in storage_lib.MOUNTABLE_STORAGE_MODES:
                 for r in self.resources:
-                    r.requires_fuse = True
+                    r.set_requires_fuse(True)
                 break
 
         return self
@@ -931,7 +932,7 @@ class Task:
             self.storage_mounts = {}
             # Clear the requires_fuse flag if no storage mounts are set.
             for r in self.resources:
-                r.requires_fuse = False
+                r.set_requires_fuse(False)
             return self
         for target, storage_obj in storage_mounts.items():
             # TODO(zhwu): /home/username/sky_workdir as the target path need
@@ -956,7 +957,7 @@ class Task:
                 # If any storage is using MOUNT mode, we need to enable FUSE in
                 # the resources.
                 for r in self.resources:
-                    r.requires_fuse = True
+                    r.set_requires_fuse(True)
         # Storage source validation is done in Storage object
         self.storage_mounts = storage_mounts
         return self
@@ -1234,13 +1235,14 @@ class Task:
 
         add_if_not_none('name', self.name)
 
-        tmp_resource_config = {}
+        tmp_resource_config: Union[Dict[str, Union[str, int]],
+                                   Dict[str, List[Dict[str, Union[str, int]]]]]
         if len(self.resources) > 1:
             resource_list = []
             for r in self.resources:
                 resource_list.append(r.to_yaml_config())
             key = 'ordered' if isinstance(self.resources, list) else 'any_of'
-            tmp_resource_config[key] = resource_list
+            tmp_resource_config = {key: resource_list}
         else:
             tmp_resource_config = list(self.resources)[0].to_yaml_config()
 
diff --git a/sky/utils/controller_utils.py b/sky/utils/controller_utils.py
index cd1788e4f..dcbf46e05 100644
--- a/sky/utils/controller_utils.py
+++ b/sky/utils/controller_utils.py
@@ -608,8 +608,9 @@ def get_controller_resources(
     controller_zone = controller_resources_to_use.zone
 
     # Filter clouds if controller_resources_to_use.cloud is specified.
-    filtered_clouds = ({controller_cloud} if controller_cloud is not None else
-                       requested_clouds_with_region_zone.keys())
+    filtered_clouds: Set[str] = {controller_cloud
+                                } if controller_cloud is not None else set(
+                                    requested_clouds_with_region_zone.keys())
 
     # Filter regions and zones and construct the result.
     result: Set[resources.Resources] = set()
@@ -618,15 +619,17 @@ def get_controller_resources(
                                                         {None: {None}})
 
         # Filter regions if controller_resources_to_use.region is specified.
-        filtered_regions = ({controller_region} if controller_region is not None
-                            else regions.keys())
+        filtered_regions: Set[Optional[str]] = ({
+            controller_region
+        } if controller_region is not None else set(regions.keys()))
 
         for region in filtered_regions:
             zones = regions.get(region, {None})
 
             # Filter zones if controller_resources_to_use.zone is specified.
-            filtered_zones = ({controller_zone}
-                              if controller_zone is not None else zones)
+            filtered_zones: Set[Optional[str]] = ({
+                controller_zone
+            } if controller_zone is not None else set(zones))
 
             # Create combinations of cloud, region, and zone.
             for zone in filtered_zones:
@@ -703,7 +706,7 @@ def _setup_proxy_command_on_controller(
     # NOTE: suppose that we have a controller in old VPC, then user
     # changes 'vpc_name' in the config and does a 'job launch' /
     # 'serve up'. In general, the old controller may not successfully
-    # launch the job in the new VPC. This happens if the two VPCs don’t
+    # launch the job in the new VPC. This happens if the two VPCs don't
     # have peering set up. Like other places in the code, we assume
     # properly setting up networking is user's responsibilities.
     # TODO(zongheng): consider adding a basic check that checks
diff --git a/sky/utils/dag_utils.py b/sky/utils/dag_utils.py
index b0099a8cb..71bb078e7 100644
--- a/sky/utils/dag_utils.py
+++ b/sky/utils/dag_utils.py
@@ -1,6 +1,6 @@
 """Utilities for loading and dumping DAGs from/to YAML files."""
 import copy
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from sky import dag as dag_lib
 from sky import sky_logging
@@ -195,7 +195,9 @@ def fill_default_config_in_dag_for_job_launch(dag: dag_lib.Dag) -> None:
         assert default_strategy is not None
         for resources in list(task_.resources):
             original_job_recovery = resources.job_recovery
-            job_recovery = {'strategy': default_strategy}
+            job_recovery: Dict[str, Optional[Union[str, int]]] = {
+                'strategy': default_strategy
+            }
             if isinstance(original_job_recovery, str):
                 job_recovery['strategy'] = original_job_recovery
             elif isinstance(original_job_recovery, dict):
diff --git a/sky/utils/resources_utils.py b/sky/utils/resources_utils.py
index 6432acd0c..60556e95b 100644
--- a/sky/utils/resources_utils.py
+++ b/sky/utils/resources_utils.py
@@ -140,10 +140,12 @@ def simplify_ports(ports: List[str]) -> List[str]:
 def format_resource(resource: 'resources_lib.Resources',
                     simplify: bool = False) -> str:
     if simplify:
+        resource = resource.assert_launchable()
         cloud = resource.cloud
         if resource.accelerators is None:
             vcpu, _ = cloud.get_vcpus_mem_from_instance_type(
                 resource.instance_type)
+            assert vcpu is not None, 'vCPU must be specified'
             hardware = f'vCPU={int(vcpu)}'
         else:
             hardware = f'{resource.accelerators}'
@@ -248,6 +250,7 @@ def make_launchables_for_valid_region_zones(
     launchables = []
     regions = launchable_resources.get_valid_regions_for_launchable()
     for region in regions:
+        assert launchable_resources.cloud is not None, 'Cloud must be specified'
         optimize_by_zone = (override_optimize_by_zone or
                             launchable_resources.cloud.optimize_by_zone())
         # It is possible that we force the optimize_by_zone but some clouds


Test patch:
diff --git a/tests/mypy_files.txt b/tests/mypy_files.txt
index df797dcb5..edd309d07 100644
--- a/tests/mypy_files.txt
+++ b/tests/mypy_files.txt
@@ -2,6 +2,4 @@ sky
 
 --exclude sky/benchmark
 --exclude sky/callbacks
---exclude sky/skylet/providers/azure
---exclude sky/resources.py
 --exclude sky/backends/monkey_patches
diff --git a/tests/unit_tests/kubernetes/test_instance_type.py b/tests/unit_tests/kubernetes/test_instance_type.py
index 563ca6f27..5a812941b 100644
--- a/tests/unit_tests/kubernetes/test_instance_type.py
+++ b/tests/unit_tests/kubernetes/test_instance_type.py
@@ -53,3 +53,19 @@ def test_kubernetes_instance_type():
             assert instance_type_from_resources.memory == memory, f'Failed from resources check for {cpus}, {memory}, {accelerator_count}, {accelerator_type}'
             assert instance_type_from_resources.accelerator_count == accelerator_count, f'Failed from resources check for {cpus}, {memory}, {accelerator_count}, {accelerator_type}'
             assert instance_type_from_resources.accelerator_type == accelerator_type, f'Failed from resources check for {cpus}, {memory}, {accelerator_count}, {accelerator_type}'
+
+    # Backward compatibility test
+    # Before https://github.com/skypilot-org/skypilot/pull/4756, the
+    # accelerators are appended with format "--{a}{type}",
+    # e.g. "4CPU--16GB--1V100".
+    # TODO(romilb): Backward compatibility, remove after 0.11.0.
+    prev_instance_type_name = '4CPU--16GB--1V100'
+    assert KubernetesInstanceType.is_valid_instance_type(
+        prev_instance_type_name
+    ), f'Failed valid instance type check for {prev_instance_type_name}'
+    cpus, memory, accelerator_count, accelerator_type = KubernetesInstanceType._parse_instance_type(
+        prev_instance_type_name)
+    assert cpus == 4, f'Failed parse check for {prev_instance_type_name}'
+    assert memory == 16, f'Failed parse check for {prev_instance_type_name}'
+    assert accelerator_count == 1, f'Failed parse check for {prev_instance_type_name}'
+    assert accelerator_type == 'V100', f'Failed parse check for {prev_instance_type_name}'