Repository: skypilot-org/skypilot
Pull request #5584 (2025-05-14 12:06:46)

Bump image tag in helm default values
<!-- Describe the changes in this PR -->

close https://github.com/skypilot-org/skypilot/issues/5503

The tag will be eventually replaced with the release version in the publish-helm workflow: https://github.com/skypilot-org/skypilot/blob/f40f78e27b076a2171f209d87db4c34758abf92b/.github/workflows/publish-helm.yml#L106

So this is no relevant to release, update the tag just to:

1. avoid confusion;
2. in case some users install the helm chart from source code

<!-- 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] Tested manually with the new bash command
- [ ] 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-build.yml b/.github/workflows/release-build.yml
index 7339f5e85..c00807c77 100644
--- a/.github/workflows/release-build.yml
+++ b/.github/workflows/release-build.yml
@@ -243,6 +243,7 @@ jobs:
           echo "Updating __version__ in sky/__init__.py and Dockerfile to ${RELEASE_VERSION}..."
           sed -i "s/__version__ = '.*'/__version__ = '${RELEASE_VERSION}'/g" sky/__init__.py
           sed -i "s/skypilot-nightly\[all\]/skypilot[all]==${RELEASE_VERSION}/g" Dockerfile
+          sed -i "s/image: berkeleyskypilot\/skypilot:.*/image: berkeleyskypilot\/skypilot:${RELEASE_VERSION}/g" charts/skypilot/values.yaml
 
           # Create the test branch from the *current* state (base branch with version bump)
           TEST_BRANCH="test_releases/${RELEASE_VERSION}"
@@ -252,6 +253,7 @@ jobs:
           # Commit the version change on the new test branch
           git add sky/__init__.py
           git add Dockerfile
+          git add charts/skypilot/values.yaml
           git commit -m "Release ${RELEASE_VERSION}"
 
           # Get the new commit SHA from the test branch
diff --git a/.github/workflows/release-publish.yml b/.github/workflows/release-publish.yml
index a6a12f64e..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}')
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/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 0ec8664df..3a1333e30 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -40,6 +40,7 @@ from sky.utils import cluster_utils
 from sky.utils import command_runner
 from sky.utils import common
 from sky.utils import common_utils
+from sky.utils import context_utils
 from sky.utils import controller_utils
 from sky.utils import env_options
 from sky.utils import registry
@@ -605,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())
@@ -1556,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:
@@ -1720,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
@@ -1739,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:
@@ -1933,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
@@ -2204,6 +2206,7 @@ def refresh_cluster_record(
 
 
 @timeline.event
+@context_utils.cancellation_guard
 def refresh_cluster_status_handle(
     cluster_name: str,
     *,
@@ -2253,6 +2256,7 @@ def check_cluster_available(
     ...
 
 
+@context_utils.cancellation_guard
 def check_cluster_available(
     cluster_name: str,
     *,
@@ -2467,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,
@@ -2965,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(
@@ -2982,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}.')
@@ -2995,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)
@@ -3005,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 {}
@@ -3014,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 e920b53bf..48eb2ac7e 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -61,6 +61,7 @@ from sky.utils import cluster_utils
 from sky.utils import command_runner
 from sky.utils import common
 from sky.utils import common_utils
+from sky.utils import context_utils
 from sky.utils import controller_utils
 from sky.utils import env_options
 from sky.utils import log_utils
@@ -274,6 +275,7 @@ class RayCodeGen:
         ray_address = 'auto'
         self._code = [
             textwrap.dedent(f"""\
+            import functools
             import getpass
             import hashlib
             import io
@@ -363,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),
@@ -1355,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)
@@ -2030,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:
@@ -2159,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
 
 
@@ -2415,6 +2422,7 @@ class CloudVmRayResourceHandle(backends.backend.ResourceHandle):
             internal_external_ips[1:], key=lambda x: x[1])
         self.stable_internal_external_ips = stable_internal_external_ips
 
+    @context_utils.cancellation_guard
     @annotations.lru_cache(scope='global')
     @timeline.event
     def get_command_runners(self,
@@ -2426,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())
@@ -3142,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(
@@ -3247,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,
@@ -3842,6 +3853,7 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
         subprocess_utils.run_in_parallel(_rsync_down, parallel_args)
         return dict(zip(job_ids, local_log_dirs))
 
+    @context_utils.cancellation_guard
     def tail_logs(self,
                   handle: CloudVmRayResourceHandle,
                   job_id: Optional[int],
@@ -4148,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
@@ -4360,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,
@@ -4489,6 +4504,7 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
 
                     # down = False is the default, but warn the user in case
                     # they have explicitly specified it.
+                    # TODO(cooperc): Fix for new autostop stuff.
                     config_override_down = skypilot_config.get_nested(
                         (controller.value.controller_type, 'controller',
                          'autostop', 'down'), None)
@@ -4559,6 +4575,7 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
     # TODO(zhwu): Refactor this to a CommandRunner class, so different backends
     # can support its own command runner.
     @timeline.event
+    @context_utils.cancellation_guard
     def run_on_head(
         self,
         handle: CloudVmRayResourceHandle,
@@ -4704,6 +4721,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..c74b1264d 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -369,9 +369,18 @@ def _start(
                 'supported when starting SkyPilot controllers. To '
                 f'fix: omit the {arguments_str} to use the '
                 f'default autostop settings from config.')
-        idle_minutes_to_autostop, down = (
-            controller_utils.get_controller_autostop_config(
-                controller=controller))
+
+        # Get the autostop resources, from which we extract the correct autostop
+        # config.
+        controller_resources = controller_utils.get_controller_resources(
+            controller, [])
+        # All resources should have the same autostop config.
+        controller_autostop_config = list(
+            controller_resources)[0].autostop_config
+        if (controller_autostop_config is not None and
+                controller_autostop_config.enabled):
+            idle_minutes_to_autostop = controller_autostop_config.idle_minutes
+            down = controller_autostop_config.down
 
     usage_lib.record_cluster_name_for_current_operation(cluster_name)
 
@@ -627,29 +636,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/execution.py b/sky/execution.py
index 0f9a0cf23..9d42ac116 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -226,11 +226,44 @@ def _execute(
     requested_features |= task.get_required_cloud_features()
 
     backend = backend if backend is not None else backends.CloudVmRayBackend()
+    # Figure out autostop config.
+    # Note: Ideally this can happen after provisioning, so we can check the
+    # autostop config from the launched resources. Before provisioning,
+    # we aren't sure which resources will be launched, and different
+    # resources may have different autostop configs.
     if isinstance(backend, backends.CloudVmRayBackend):
         if down and idle_minutes_to_autostop is None:
             # Use auto{stop,down} to terminate the cluster after the task is
             # done.
             idle_minutes_to_autostop = 0
+        elif not down and idle_minutes_to_autostop is None:
+            # No autostop config specified on command line, use the
+            # config from resources.
+            # TODO(cooperc): This should be done after provisioning, in order to
+            # support different autostop configs for different resources.
+            # Blockers:
+            # - Need autostop config to set requested_features before
+            #   provisioning.
+            # - Need to send info message about idle_minutes_to_autostop==0 here
+            # - Need to check if autostop is supported by the backend.
+            resources = list(task.resources)
+            for resource in resources:
+                if resource.autostop_config != resources[0].autostop_config:
+                    raise ValueError(
+                        'All resources must have the same autostop config.')
+            resource_autostop_config = resources[0].autostop_config
+
+            if resource_autostop_config is not None:
+                if resource_autostop_config.enabled:
+                    idle_minutes_to_autostop = (
+                        resource_autostop_config.idle_minutes)
+                    down = resource_autostop_config.down
+                else:
+                    # Autostop is explicitly disabled, so cancel it if it's
+                    # already set.
+                    assert not resource_autostop_config.enabled
+                    idle_minutes_to_autostop = -1
+                    down = False
         if idle_minutes_to_autostop is not None:
             if idle_minutes_to_autostop == 0:
                 # idle_minutes_to_autostop=0 can cause the following problem:
diff --git a/sky/global_user_state.py b/sky/global_user_state.py
index e1c9ce21d..6dc1bb6ad 100644
--- a/sky/global_user_state.py
+++ b/sky/global_user_state.py
@@ -19,6 +19,7 @@ import uuid
 from sky import models
 from sky import sky_logging
 from sky.utils import common_utils
+from sky.utils import context_utils
 from sky.utils import db_utils
 from sky.utils import registry
 from sky.utils import status_lib
@@ -671,6 +672,7 @@ def _load_storage_mounts_metadata(
     return pickle.loads(record_storage_mounts_metadata)
 
 
+@context_utils.cancellation_guard
 def get_cluster_from_name(
         cluster_name: Optional[str]) -> Optional[Dict[str, Any]]:
     rows = _DB.cursor.execute(
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/jobs/server/core.py b/sky/jobs/server/core.py
index dc178fe27..09080c8a0 100644
--- a/sky/jobs/server/core.py
+++ b/sky/jobs/server/core.py
@@ -172,9 +172,6 @@ def launch(
         controller_resources = controller_utils.get_controller_resources(
             controller=controller,
             task_resources=sum([list(t.resources) for t in dag.tasks], []))
-        controller_idle_minutes_to_autostop, controller_down = (
-            controller_utils.get_controller_autostop_config(
-                controller=controller))
 
         vars_to_fill = {
             'remote_user_yaml_path': remote_user_yaml_path,
@@ -216,15 +213,12 @@ def launch(
         # Launch with the api server's user hash, so that sky status does not
         # show the owner of the controller as whatever user launched it first.
         with common.with_server_user_hash():
-            return execution.launch(
-                task=controller_task,
-                cluster_name=controller_name,
-                stream_logs=stream_logs,
-                idle_minutes_to_autostop=controller_idle_minutes_to_autostop,
-                down=controller_down,
-                retry_until_up=True,
-                fast=True,
-                _disable_controller_check=True)
+            return execution.launch(task=controller_task,
+                                    cluster_name=controller_name,
+                                    stream_logs=stream_logs,
+                                    retry_until_up=True,
+                                    fast=True,
+                                    _disable_controller_check=True)
 
 
 def queue_from_kubernetes_pod(
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..f988c0005 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -1,7 +1,8 @@
 """Resources: compute requirements of Tasks."""
 import dataclasses
 import textwrap
-from typing import Any, Dict, List, Optional, Set, Tuple, Union
+import typing
+from typing import Any, Dict, List, Literal, Optional, Set, Tuple, Union
 
 import colorama
 
@@ -34,6 +35,50 @@ RESOURCE_CONFIG_ALIASES = {
 }
 
 
+@dataclasses.dataclass
+class AutostopConfig:
+    """Configuration for autostop."""
+    # enabled isn't present in the yaml config, but it's needed for this class
+    # to be complete.
+    enabled: bool
+    # If enabled is False, these values are ignored.
+    idle_minutes: int = 5
+    down: bool = False
+
+    def to_yaml_config(self) -> Union[Literal[False], Dict[str, Any]]:
+        if not self.enabled:
+            return False
+        return {
+            'idle_minutes': self.idle_minutes,
+            'down': self.down,
+        }
+
+    @classmethod
+    def from_yaml_config(
+        cls, config: Union[bool, int, Dict[str, Any], None]
+    ) -> Optional['AutostopConfig']:
+        if isinstance(config, bool):
+            if config:
+                return cls(enabled=True)
+            else:
+                return cls(enabled=False)
+
+        if isinstance(config, int):
+            return cls(idle_minutes=config, down=False, enabled=True)
+
+        if isinstance(config, dict):
+            # If we have a dict, autostop is enabled. (Only way to disable is
+            # with `false`, a bool.)
+            autostop_config = cls(enabled=True)
+            if 'idle_minutes' in config:
+                autostop_config.idle_minutes = config['idle_minutes']
+            if 'down' in config:
+                autostop_config.down = config['down']
+            return autostop_config
+
+        return None
+
+
 class Resources:
     """Resources: compute requirements of Tasks.
 
@@ -51,7 +96,7 @@ class Resources:
     """
     # If any fields changed, increment the version. For backward compatibility,
     # modify the __setstate__ method to handle the old version.
-    _VERSION = 22
+    _VERSION = 23
 
     def __init__(
         self,
@@ -59,17 +104,19 @@ 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,
         labels: Optional[Dict[str, str]] = None,
+        autostop: Union[bool, int, Dict[str, Any], None] = None,
         # Internal use only.
         # pylint: disable=invalid-name
         _docker_login_config: Optional[docker_utils.DockerLoginConfig] = None,
@@ -152,6 +199,8 @@ class Resources:
             instance tags. On GCP, labels map to instance labels. On
             Kubernetes, labels map to pod labels. On other clouds, labels are
             not supported and will be ignored.
+          autostop: the autostop configuration to use. For launched resources,
+            may or may not correspond to the actual current autostop config.
           _docker_login_config: the docker configuration to use. This includes
             the docker username, password, and registry server. If None, skip
             docker login.
@@ -177,7 +226,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 +238,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 +251,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 +259,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 +281,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,11 +303,12 @@ 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)
         self._set_accelerators(accelerators, accelerator_args)
+        self._set_autostop_config(autostop)
 
     def validate(self):
         """Validate the resources and infer the missing fields if possible."""
@@ -378,19 +432,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 +498,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 +510,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 +518,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
@@ -479,6 +533,16 @@ class Resources:
     def labels(self) -> Optional[Dict[str, str]]:
         return self._labels
 
+    @property
+    def autostop_config(self) -> Optional[AutostopConfig]:
+        """The requested autostop config.
+
+        Warning: This is the autostop config that was originally used to
+        launch the resources. It may not correspond to the actual current
+        autostop config.
+        """
+        return self._autostop_config
+
     @property
     def is_image_managed(self) -> Optional[bool]:
         return self._is_image_managed
@@ -489,16 +553,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 +642,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 +678,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 +716,34 @@ 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 _set_autostop_config(
+        self,
+        autostop: Union[bool, int, Dict[str, Any], None],
+    ) -> None:
+        self._autostop_config = AutostopConfig.from_yaml_config(autostop)
 
     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 +800,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 +835,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 +941,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 +959,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 +986,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 +1022,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 +1045,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 +1157,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 +1204,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 +1259,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 +1284,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 +1375,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 +1414,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(
@@ -1314,6 +1427,10 @@ class Resources:
             if elem is not None:
                 override_configs.set_nested(key, elem)
 
+        current_autostop_config = None
+        if self.autostop_config is not None:
+            current_autostop_config = self.autostop_config.to_yaml_config()
+
         override_configs = dict(override_configs) if override_configs else None
         resources = Resources(
             cloud=override.pop('cloud', self.cloud),
@@ -1332,6 +1449,7 @@ class Resources:
             disk_tier=override.pop('disk_tier', self.disk_tier),
             ports=override.pop('ports', self.ports),
             labels=override.pop('labels', self.labels),
+            autostop=override.pop('autostop', current_autostop_config),
             _docker_login_config=override.pop('_docker_login_config',
                                               self._docker_login_config),
             _docker_username_for_runpod=override.pop(
@@ -1529,6 +1647,7 @@ class Resources:
         resources_fields['disk_tier'] = config.pop('disk_tier', None)
         resources_fields['ports'] = config.pop('ports', None)
         resources_fields['labels'] = config.pop('labels', None)
+        resources_fields['autostop'] = config.pop('autostop', None)
         resources_fields['_docker_login_config'] = config.pop(
             '_docker_login_config', None)
         resources_fields['_docker_username_for_runpod'] = config.pop(
@@ -1578,6 +1697,8 @@ class Resources:
             config['disk_tier'] = self.disk_tier.value
         add_if_not_none('ports', self.ports)
         add_if_not_none('labels', self.labels)
+        if self._autostop_config is not None:
+            config['autostop'] = self._autostop_config.to_yaml_config()
         if self._docker_login_config is not None:
             config['_docker_login_config'] = dataclasses.asdict(
                 self._docker_login_config)
@@ -1733,4 +1854,42 @@ class Resources:
             self._docker_username_for_runpod = state.pop(
                 '_docker_username_for_runpod', None)
 
+        if version < 23:
+            self._autostop_config = 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/server/core.py b/sky/serve/server/core.py
index 1b66bdbe1..7aeaa0272 100644
--- a/sky/serve/server/core.py
+++ b/sky/serve/server/core.py
@@ -219,17 +219,12 @@ def up(
         # whether the service is already running. If the id is the same
         # with the current job id, we know the service is up and running
         # for the first time; otherwise it is a name conflict.
-        controller_idle_minutes_to_autostop, controller_down = (
-            controller_utils.get_controller_autostop_config(
-                controller=controller_utils.Controllers.SKY_SERVE_CONTROLLER))
         # Since the controller may be shared among multiple users, launch the
         # controller with the API server's user hash.
         with common.with_server_user_hash():
             controller_job_id, controller_handle = execution.launch(
                 task=controller_task,
                 cluster_name=controller_name,
-                idle_minutes_to_autostop=controller_idle_minutes_to_autostop,
-                down=controller_down,
                 retry_until_up=True,
                 _disable_controller_check=True,
             )
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/requests/executor.py b/sky/server/requests/executor.py
index 5317856bd..2cd019dd4 100644
--- a/sky/server/requests/executor.py
+++ b/sky/server/requests/executor.py
@@ -18,7 +18,10 @@ The number of the workers is determined by the system resources.
 
 See the [README.md](../README.md) for detailed architecture of the executor.
 """
+import asyncio
 import contextlib
+import contextvars
+import functools
 import multiprocessing
 import os
 import queue as queue_lib
@@ -47,6 +50,7 @@ from sky.server.requests.queues import mp_queue
 from sky.skylet import constants
 from sky.utils import annotations
 from sky.utils import common_utils
+from sky.utils import context
 from sky.utils import subprocess_utils
 from sky.utils import timeline
 
@@ -60,7 +64,6 @@ else:
     from typing_extensions import ParamSpec
 
 P = ParamSpec('P')
-
 logger = sky_logging.init_logger(__name__)
 
 # On macOS, the default start method for multiprocessing is 'fork', which
@@ -341,6 +344,114 @@ def _request_execution_wrapper(request_id: str,
             logger.info(f'Request {request_id} finished')
 
 
+async def execute_request_coroutine(request: api_requests.Request):
+    """Execute a request in current event loop.
+
+    Similar to _request_execution_wrapper, but executed as coroutine in current
+    event loop. This is designed for executing tasks that are not CPU
+    intensive, e.g. sky logs.
+    """
+    ctx = context.get()
+    if ctx is None:
+        raise ValueError('Context is not initialized')
+    logger.info(f'Executing request {request.request_id} in coroutine')
+    func = request.entrypoint
+    request_body = request.request_body
+    with api_requests.update_request(request.request_id) as request_task:
+        request_task.status = api_requests.RequestStatus.RUNNING
+    # Redirect stdout and stderr to the request log path.
+    original_output = ctx.redirect_log(request.log_path)
+    # Override environment variables that backs env_options.Options
+    # TODO(aylei): compared to process executor, running task in coroutine has
+    # two issues to fix:
+    # 1. skypilot config is not contextual
+    # 2. envs that read directly from os.environ are not contextual
+    ctx.override_envs(request_body.env_vars)
+    loop = asyncio.get_running_loop()
+    pyctx = contextvars.copy_context()
+    func_call = functools.partial(pyctx.run, func, **request_body.to_kwargs())
+    fut: asyncio.Future = loop.run_in_executor(None, func_call)
+
+    async def poll_task(request_id: str) -> bool:
+        request = api_requests.get_request(request_id)
+        if request is None:
+            raise RuntimeError('Request not found')
+
+        if request.status == api_requests.RequestStatus.CANCELLED:
+            ctx.cancel()
+            return True
+
+        if fut.done():
+            try:
+                result = await fut
+                api_requests.set_request_succeeded(request_id, result)
+            except asyncio.CancelledError:
+                # The task is cancelled by ctx.cancel(), where the status
+                # should already be set to CANCELLED.
+                pass
+            except Exception as e:  # pylint: disable=broad-except
+                ctx.redirect_log(original_output)
+                api_requests.set_request_failed(request_id, e)
+                logger.error(f'Request {request_id} failed due to '
+                             f'{common_utils.format_exception(e)}')
+            return True
+        return False
+
+    try:
+        while True:
+            res = await poll_task(request.request_id)
+            if res:
+                break
+            await asyncio.sleep(0.5)
+    except asyncio.CancelledError:
+        # Current coroutine is cancelled due to client disconnect, set the
+        # request status for consistency.
+        api_requests.set_request_cancelled(request.request_id)
+        pass
+    # pylint: disable=broad-except
+    except (Exception, KeyboardInterrupt, SystemExit) as e:
+        # Handle any other error
+        ctx.redirect_log(original_output)
+        ctx.cancel()
+        api_requests.set_request_failed(request.request_id, e)
+        logger.error(f'Request {request.request_id} interrupted due to '
+                     f'unhandled exception: {common_utils.format_exception(e)}')
+        raise
+
+
+def prepare_request(
+    request_id: str,
+    request_name: str,
+    request_body: payloads.RequestBody,
+    func: Callable[P, Any],
+    request_cluster_name: Optional[str] = None,
+    schedule_type: api_requests.ScheduleType = (api_requests.ScheduleType.LONG),
+    is_skypilot_system: bool = False,
+) -> api_requests.Request:
+    """Prepare a request for execution."""
+    user_id = request_body.env_vars[constants.USER_ID_ENV_VAR]
+    if is_skypilot_system:
+        user_id = server_constants.SKYPILOT_SYSTEM_USER_ID
+        global_user_state.add_or_update_user(
+            models.User(id=user_id, name=user_id))
+    request = api_requests.Request(request_id=request_id,
+                                   name=server_constants.REQUEST_NAME_PREFIX +
+                                   request_name,
+                                   entrypoint=func,
+                                   request_body=request_body,
+                                   status=api_requests.RequestStatus.PENDING,
+                                   created_at=time.time(),
+                                   schedule_type=schedule_type,
+                                   user_id=user_id,
+                                   cluster_name=request_cluster_name)
+
+    if not api_requests.create_if_not_exists(request):
+        raise RuntimeError(f'Request {request_id} already exists.')
+
+    request.log_path.touch()
+    return request
+
+
 def schedule_request(
         request_id: str,
         request_name: str,
@@ -372,27 +483,8 @@ def schedule_request(
             The precondition is waited asynchronously and does not block the
             caller.
     """
-    user_id = request_body.env_vars[constants.USER_ID_ENV_VAR]
-    if is_skypilot_system:
-        user_id = server_constants.SKYPILOT_SYSTEM_USER_ID
-        global_user_state.add_or_update_user(
-            models.User(id=user_id, name=user_id))
-    request = api_requests.Request(request_id=request_id,
-                                   name=server_constants.REQUEST_NAME_PREFIX +
-                                   request_name,
-                                   entrypoint=func,
-                                   request_body=request_body,
-                                   status=api_requests.RequestStatus.PENDING,
-                                   created_at=time.time(),
-                                   schedule_type=schedule_type,
-                                   user_id=user_id,
-                                   cluster_name=request_cluster_name)
-
-    if not api_requests.create_if_not_exists(request):
-        logger.debug(f'Request {request_id} already exists.')
-        return
-
-    request.log_path.touch()
+    prepare_request(request_id, request_name, request_body, func,
+                    request_cluster_name, schedule_type, is_skypilot_system)
 
     def enqueue():
         input_tuple = (request_id, ignore_return_value)
diff --git a/sky/server/requests/requests.py b/sky/server/requests/requests.py
index d417e0295..cce2c0138 100644
--- a/sky/server/requests/requests.py
+++ b/sky/server/requests/requests.py
@@ -606,3 +606,18 @@ def set_request_failed(request_id: str, e: BaseException) -> None:
         assert request_task is not None, request_id
         request_task.status = RequestStatus.FAILED
         request_task.set_error(e)
+
+
+def set_request_succeeded(request_id: str, result: Any) -> None:
+    """Set a request to succeeded and populate the result."""
+    with update_request(request_id) as request_task:
+        assert request_task is not None, request_id
+        request_task.status = RequestStatus.SUCCEEDED
+        request_task.set_return_value(result)
+
+
+def set_request_cancelled(request_id: str) -> None:
+    """Set a request to cancelled."""
+    with update_request(request_id) as request_task:
+        assert request_task is not None, request_id
+        request_task.status = RequestStatus.CANCELLED
diff --git a/sky/server/server.py b/sky/server/server.py
index 7c0f0fc06..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
@@ -47,6 +48,7 @@ from sky.usage import usage_lib
 from sky.utils import admin_policy_utils
 from sky.utils import common as common_lib
 from sky.utils import common_utils
+from sky.utils import context
 from sky.utils import dag_utils
 from sky.utils import env_options
 from sky.utils import status_lib
@@ -166,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
@@ -673,24 +703,28 @@ async def logs(
     # TODO(zhwu): This should wait for the request on the cluster, e.g., async
     # launch, to finish, so that a user does not need to manually pull the
     # request status.
-    executor.schedule_request(
+    # Only initialize the context in logs handler to limit the scope of this
+    # experimental change.
+    # TODO(aylei): init in lifespan() to enable SkyPilot context in all APIs.
+    context.initialize()
+    request_task = executor.prepare_request(
         request_id=request.state.request_id,
         request_name='logs',
         request_body=cluster_job_body,
         func=core.tail_logs,
-        # TODO(aylei): We have tail logs scheduled as SHORT request, because it
-        # should be responsive. However, it can be long running if the user's
-        # job keeps running, and we should avoid it taking the SHORT worker.
         schedule_type=requests_lib.ScheduleType.SHORT,
-        request_cluster_name=cluster_job_body.cluster_name,
     )
+    task = asyncio.create_task(executor.execute_request_coroutine(request_task))
 
-    request_task = requests_lib.get_request(request.state.request_id)
+    def cancel_task():
+        task.cancel()
 
+    # Cancel the task after the request is done or client disconnects
+    background_tasks.add_task(cancel_task)
     # TODO(zhwu): This makes viewing logs in browser impossible. We should adopt
     # the same approach as /stream.
     return stream_utils.stream_response(
-        request_id=request_task.request_id,
+        request_id=request.state.request_id,
         logs_path=request_task.log_path,
         background_tasks=background_tasks,
     )
@@ -1125,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
 
-    # Try to serve the file directly from the out directory first
+    Returns:
+        FileResponse for static files or index.html for client-side routing.
+
+    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:
@@ -1154,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/server/uvicorn.py b/sky/server/uvicorn.py
index 5ad7db485..e99213ea6 100644
--- a/sky/server/uvicorn.py
+++ b/sky/server/uvicorn.py
@@ -3,6 +3,7 @@
 This module is a wrapper around uvicorn to customize the behavior of the
 server.
 """
+import functools
 import os
 import threading
 from typing import Optional
@@ -10,6 +11,7 @@ from typing import Optional
 import uvicorn
 from uvicorn.supervisors import multiprocess
 
+from sky.utils import context_utils
 from sky.utils import subprocess_utils
 
 
@@ -21,19 +23,27 @@ def run(config: uvicorn.Config):
         # guard by an exception.
         raise ValueError('Reload is not supported yet.')
     server = uvicorn.Server(config=config)
+    run_server_process = functools.partial(_run_server_process, server)
     try:
         if config.workers is not None and config.workers > 1:
             sock = config.bind_socket()
-            SlowStartMultiprocess(config, target=server.run,
+            SlowStartMultiprocess(config,
+                                  target=run_server_process,
                                   sockets=[sock]).run()
         else:
-            server.run()
+            run_server_process()
     finally:
         # Copied from unvicorn.run()
         if config.uds and os.path.exists(config.uds):
             os.remove(config.uds)
 
 
+def _run_server_process(server: uvicorn.Server, *args, **kwargs):
+    """Run the server process with contextually aware."""
+    context_utils.hijack_sys_attrs()
+    server.run(*args, **kwargs)
+
+
 class SlowStartMultiprocess(multiprocess.Multiprocess):
     """Uvicorn Multiprocess wrapper with slow start.
 
diff --git a/sky/sky_logging.py b/sky/sky_logging.py
index 7021ec9bc..f38147dff 100644
--- a/sky/sky_logging.py
+++ b/sky/sky_logging.py
@@ -10,6 +10,7 @@ import threading
 import colorama
 
 from sky.skylet import constants
+from sky.utils import context
 from sky.utils import env_options
 from sky.utils import rich_utils
 
@@ -47,6 +48,43 @@ class NewLineFormatter(logging.Formatter):
         return msg
 
 
+class EnvAwareHandler(rich_utils.RichSafeStreamHandler):
+    """A handler that awares environment variables.
+
+    This handler dynamically reflects the log level from environment variables.
+    """
+
+    def __init__(self, stream=None, level=logging.NOTSET, sensitive=False):
+        super().__init__(stream)
+        self.level = level
+        self._sensitive = sensitive
+
+    @property
+    def level(self):
+        # Only refresh log level if we are in a context, since the log level
+        # has already been reloaded eagerly in multi-processing. Refresh again
+        # is a no-op and can be avoided.
+        # TODO(aylei): unify the mechanism for coroutine context and
+        # multi-processing.
+        if context.get() is not None:
+            if self._sensitive:
+                # For sensitive logger, suppress debug log despite the
+                # SKYPILOT_DEBUG env var if SUPPRESS_SENSITIVE_LOG is set
+                if env_options.Options.SUPPRESS_SENSITIVE_LOG.get():
+                    return logging.INFO
+            if env_options.Options.SHOW_DEBUG_INFO.get():
+                return logging.DEBUG
+            else:
+                return self._level
+        else:
+            return self._level
+
+    @level.setter
+    def level(self, level):
+        # pylint: disable=protected-access
+        self._level = logging._checkLevel(level)
+
+
 _root_logger = logging.getLogger('sky')
 _default_handler = None
 _logging_config = threading.local()
@@ -67,7 +105,7 @@ def _setup_logger():
     _root_logger.setLevel(logging.DEBUG)
     global _default_handler
     if _default_handler is None:
-        _default_handler = rich_utils.RichSafeStreamHandler(sys.stdout)
+        _default_handler = EnvAwareHandler(sys.stdout)
         _default_handler.flush = sys.stdout.flush  # type: ignore
         if env_options.Options.SHOW_DEBUG_INFO.get():
             _default_handler.setLevel(logging.DEBUG)
@@ -87,7 +125,7 @@ def _setup_logger():
         # for certain loggers.
         for logger_name in _SENSITIVE_LOGGER:
             logger = logging.getLogger(logger_name)
-            handler_to_logger = rich_utils.RichSafeStreamHandler(sys.stdout)
+            handler_to_logger = EnvAwareHandler(sys.stdout, sensitive=True)
             handler_to_logger.flush = sys.stdout.flush  # type: ignore
             logger.addHandler(handler_to_logger)
             logger.setLevel(logging.INFO)
diff --git a/sky/skylet/log_lib.py b/sky/skylet/log_lib.py
index 24df608cd..878ee26a4 100644
--- a/sky/skylet/log_lib.py
+++ b/sky/skylet/log_lib.py
@@ -4,6 +4,7 @@ This is a remote utility module that provides logging functionality.
 """
 import collections
 import copy
+import functools
 import io
 import multiprocessing.pool
 import os
@@ -21,6 +22,8 @@ import colorama
 from sky import sky_logging
 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
 from sky.utils import ux_utils
@@ -59,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,
@@ -77,6 +90,9 @@ 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 = _get_context()
+                if ctx is not None and ctx.is_canceled():
+                    return
                 line = out_io.readline()
                 if not line:
                     break
@@ -111,26 +127,24 @@ def _handle_io_stream(io_stream, out_stream, args: _ProcessingArgs):
     return ''.join(out)
 
 
-def process_subprocess_stream(proc, args: _ProcessingArgs) -> Tuple[str, str]:
-    """Redirect the process's filtered stdout/stderr to both stream and file"""
+def process_subprocess_stream(proc, stdout_stream_handler,
+                              stderr_stream_handler) -> Tuple[str, str]:
+    """Process the stream of a process in threads, blocking."""
     if proc.stderr is not None:
         # Asyncio does not work as the output processing can be executed in a
         # different thread.
         # selectors is possible to handle the multiplexing of stdout/stderr,
         # but it introduces buffering making the output not streaming.
         with multiprocessing.pool.ThreadPool(processes=1) as pool:
-            err_args = copy.copy(args)
-            err_args.line_processor = None
-            stderr_fut = pool.apply_async(_handle_io_stream,
-                                          args=(proc.stderr, sys.stderr,
-                                                err_args))
+            stderr_fut = pool.apply_async(stderr_stream_handler,
+                                          args=(proc.stderr, sys.stderr))
             # Do not launch a thread for stdout as the rich.status does not
             # work in a thread, which is used in
             # log_utils.RayUpLineProcessor.
-            stdout = _handle_io_stream(proc.stdout, sys.stdout, args)
+            stdout = stdout_stream_handler(proc.stdout, sys.stdout)
             stderr = stderr_fut.get()
     else:
-        stdout = _handle_io_stream(proc.stdout, sys.stdout, args)
+        stdout = stdout_stream_handler(proc.stdout, sys.stdout)
         stderr = ''
     return stdout, stderr
 
@@ -176,7 +190,12 @@ def run_with_log(
     # Redirect stderr to stdout when using ray, to preserve the order of
     # stdout and stderr.
     stdout_arg = stderr_arg = None
-    if process_stream:
+    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)
+        # 2. Potential contextual handling is needed (ctx is not None)
+        # TODO(aylei): can we always capture the stdout/stderr?
         stdout_arg = subprocess.PIPE
         stderr_arg = subprocess.PIPE if not with_ray else subprocess.STDOUT
     # Use stdin=subprocess.DEVNULL by default, as allowing inputs will mess up
@@ -197,6 +216,8 @@ def run_with_log(
             subprocess_utils.kill_process_daemon(proc.pid)
             stdout = ''
             stderr = ''
+            stdout_stream_handler = None
+            stderr_stream_handler = None
 
             if process_stream:
                 if skip_lines is None:
@@ -223,7 +244,35 @@ def run_with_log(
                     replace_crlf=with_ray,
                     streaming_prefix=streaming_prefix,
                 )
-                stdout, stderr = process_subprocess_stream(proc, args)
+                stdout_stream_handler = functools.partial(
+                    _handle_io_stream,
+                    args=args,
+                )
+                if proc.stderr is not None:
+                    err_args = copy.copy(args)
+                    err_args.line_processor = None
+                    stderr_stream_handler = functools.partial(
+                        _handle_io_stream,
+                        args=err_args,
+                    )
+            if ctx is not None:
+                # When runs in a coroutine, always process the subprocess
+                # stream to:
+                # 1. handle context cancellation
+                # 2. redirect subprocess stdout/stderr to the contextual
+                #    stdout/stderr of current coroutine.
+                stdout, stderr = context_utils.pipe_and_wait_process(
+                    ctx,
+                    proc,
+                    cancel_callback=subprocess_utils.kill_children_processes,
+                    stdout_stream_handler=stdout_stream_handler,
+                    stderr_stream_handler=stderr_stream_handler)
+            elif process_stream:
+                # When runs in a process, only process subprocess stream if
+                # necessary to avoid unnecessary stream handling overhead.
+                stdout, stderr = process_subprocess_stream(
+                    proc, stdout_stream_handler, stderr_stream_handler)
+            # Ensure returncode is set.
             proc.wait()
             if require_outputs:
                 return proc.returncode, stdout, stderr
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/cli_utils/status_utils.py b/sky/utils/cli_utils/status_utils.py
index f6ea56ca8..6e267770b 100644
--- a/sky/utils/cli_utils/status_utils.py
+++ b/sky/utils/cli_utils/status_utils.py
@@ -201,12 +201,13 @@ def show_cost_report_table(cluster_records: List[_ClusterCostReportRecord],
             controller = controller_utils.Controllers.from_name(controller_name)
             if controller is None:
                 raise ValueError(f'Controller {controller_name} not found.')
-            autostop_minutes, _ = (
-                controller_utils.get_controller_autostop_config(
-                    controller=controller))
-            if autostop_minutes is not None:
+            controller_handle: backends.CloudVmRayResourceHandle = (
+                cluster_records[0]['handle'])
+            autostop_config = (
+                controller_handle.launched_resources.autostop_config)
+            if autostop_config is not None:
                 autostop_str = (f'{colorama.Style.DIM} (will be autostopped if '
-                                f'idle for {autostop_minutes}min)'
+                                f'idle for {autostop_config.idle_minutes}min)'
                                 f'{colorama.Style.RESET_ALL}')
             click.echo(f'\n{colorama.Fore.CYAN}{colorama.Style.BRIGHT}'
                        f'{controller_name}{colorama.Style.RESET_ALL}'
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index d1bdd9ab6..b6e2bbb9e 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -11,6 +11,7 @@ from sky import sky_logging
 from sky.skylet import constants
 from sky.skylet import log_lib
 from sky.utils import common_utils
+from sky.utils import context_utils
 from sky.utils import control_master_utils
 from sky.utils import subprocess_utils
 from sky.utils import timeline
@@ -574,6 +575,7 @@ class SSHCommandRunner(CommandRunner):
                                      shell=True)
 
     @timeline.event
+    @context_utils.cancellation_guard
     def run(
             self,
             cmd: Union[str, List[str]],
@@ -779,6 +781,7 @@ class KubernetesCommandRunner(CommandRunner):
         return kubectl_cmd
 
     @timeline.event
+    @context_utils.cancellation_guard
     def run(
             self,
             cmd: Union[str, List[str]],
diff --git a/sky/utils/context.py b/sky/utils/context.py
new file mode 100644
index 000000000..9376e88de
--- /dev/null
+++ b/sky/utils/context.py
@@ -0,0 +1,264 @@
+"""SkyPilot context for threads and coroutines."""
+
+import asyncio
+from collections.abc import Mapping
+from collections.abc import MutableMapping
+import contextvars
+import os
+import pathlib
+import subprocess
+import sys
+from typing import Dict, Optional, TextIO
+
+
+class Context(object):
+    """SkyPilot typed context vars for threads and coroutines.
+
+    This is a wrapper around `contextvars.ContextVar` that provides a typed
+    interface for the SkyPilot specific context variables that can be accessed
+    at any layer of the call stack. ContextVar is coroutine local, an empty
+    Context will be intialized for each coroutine when it is created.
+
+    Adding a new context variable for a new feature is as simple as:
+    1. Add a new instance variable to the Context class.
+    2. (Optional) Add new accessor methods if the variable should be protected.
+
+    To propagate the context to a new thread/coroutine, use
+    `contextvars.copy_context()`.
+
+    Example:
+        import asyncio
+        import contextvars
+        import time
+        from sky.utils import context
+
+        def sync_task():
+            while True:
+                if context.get().is_canceled():
+                    break
+                time.sleep(1)
+
+        async def fastapi_handler():
+            # context.initialize() has been called in lifespan
+            ctx = contextvars.copy_context()
+            # asyncio.to_thread copies current context implicitly
+            task = asyncio.to_thread(sync_task)
+            # Or explicitly:
+            # loop = asyncio.get_running_loop()
+            # ctx = contextvars.copy_context()
+            # task = loop.run_in_executor(None, ctx.run, sync_task)
+            await asyncio.sleep(1)
+            context.get().cancel()
+            await task
+    """
+
+    def __init__(self):
+        self._canceled = asyncio.Event()
+        self._log_file = None
+        self._log_file_handle = None
+        self.env_overrides = {}
+
+    def cancel(self):
+        """Cancel the context."""
+        self._canceled.set()
+
+    def is_canceled(self):
+        """Check if the context is canceled."""
+        return self._canceled.is_set()
+
+    def redirect_log(
+            self, log_file: Optional[pathlib.Path]) -> Optional[pathlib.Path]:
+        """Redirect the stdout and stderr of current context to a file.
+
+        Args:
+            log_file: The log file to redirect to. If None, the stdout and
+            stderr will be restored to the original streams.
+
+        Returns:
+            The old log file, or None if the stdout and stderr were not
+            redirected.
+        """
+        original_log_file = self._log_file
+        original_log_handle = self._log_file_handle
+        if log_file is None:
+            self._log_file_handle = None
+        else:
+            self._log_file_handle = open(log_file, 'a', encoding='utf-8')
+        self._log_file = log_file
+        if original_log_file is not None:
+            original_log_handle.close()
+        return original_log_file
+
+    def output_stream(self, fallback: TextIO) -> TextIO:
+        if self._log_file_handle is None:
+            return fallback
+        else:
+            return self._log_file_handle
+
+    def override_envs(self, envs: Dict[str, str]):
+        for k, v in envs.items():
+            self.env_overrides[k] = v
+
+
+_CONTEXT = contextvars.ContextVar('sky_context', default=None)
+
+
+def get() -> Optional[Context]:
+    """Get the current SkyPilot context.
+
+    If the context is not initialized, get() will return None. This helps
+    sync code to check whether it runs in a cancellable context and avoid
+    polling the cancellation event if it is not.
+    """
+    return _CONTEXT.get()
+
+
+class ContextualEnviron(MutableMapping):
+    """Environment variables wrapper with contextual overrides.
+
+    An instance of ContextualEnviron will typically be used to replace
+    os.environ to make the envron access of current process contextual
+    aware.
+
+    Behavior of spawning a subprocess:
+    - The contexual overrides will not be applied to the subprocess by
+      default.
+    - When using env=os.environ to pass the environment variables to the
+      subprocess explicitly. The subprocess will inherit the contextual
+      environment variables at the time of the spawn, that is, it will not
+      see the updates to the environment variables after the spawn. Also,
+      os.environ of the subprocess will not be a ContextualEnviron unless
+      the subprocess hijacks os.environ explicitly.
+    - Optionally, context.Popen() can be used to automatically pass
+      os.environ with overrides to subprocess.
+
+
+    Example:
+    1. Parent process:
+       # Hijack os.environ to be a ContextualEnviron
+       os.environ = ContextualEnviron(os.environ)
+       ctx = context.get()
+       ctx.override_envs({'FOO': 'BAR1'})
+       proc = subprocess.Popen(..., env=os.environ)
+       # Or use context.Popen instead
+       # proc = context.Popen(...)
+       ctx.override_envs({'FOO': 'BAR2'})
+    2. Subprocess:
+       assert os.environ['FOO'] == 'BAR1'
+       ctx = context.get()
+       # Override the contextual env var in the subprocess does not take
+       # effect since the os.environ is not hijacked.
+       ctx.override_envs({'FOO': 'BAR3'})
+       assert os.environ['FOO'] == 'BAR1'
+    """
+
+    def __init__(self, environ):
+        self._environ = environ
+
+    def __getitem__(self, key):
+        ctx = get()
+        if ctx is not None:
+            if key in ctx.env_overrides:
+                return ctx.env_overrides[key]
+        return self._environ[key]
+
+    def __iter__(self):
+        ctx = get()
+        if ctx is not None:
+            for key in ctx.env_overrides:
+                yield key
+            for key in self._environ:
+                # Deduplicate the keys
+                if key not in ctx.env_overrides:
+                    yield key
+        else:
+            return self._environ.__iter__()
+
+    def __len__(self):
+        return len(dict(self))
+
+    def __setitem__(self, key, value):
+        return self._environ.__setitem__(key, value)
+
+    def __delitem__(self, key):
+        return self._environ.__delitem__(key)
+
+    def __repr__(self):
+        return self._environ.__repr__()
+
+    def copy(self):
+        copied = self._environ.copy()
+        ctx = get()
+        if ctx is not None:
+            copied.update(ctx.env_overrides)
+        return copied
+
+    def setdefault(self, key, default=None):
+        return self._environ.setdefault(key, default)
+
+    def __ior__(self, other):
+        if not isinstance(other, Mapping):
+            return NotImplemented
+        self.update(other)
+        return self
+
+    def __or__(self, other):
+        if not isinstance(other, Mapping):
+            return NotImplemented
+        new = dict(self)
+        new.update(other)
+        return new
+
+    def __ror__(self, other):
+        if not isinstance(other, Mapping):
+            return NotImplemented
+        new = dict(other)
+        new.update(self)
+        return new
+
+
+class Popen(subprocess.Popen):
+
+    def __init__(self, *args, **kwargs):
+        env = kwargs.pop('env', None)
+        if env is None:
+            env = os.environ
+        super().__init__(*args, env=env, **kwargs)
+
+
+def initialize():
+    """Initialize the current SkyPilot context."""
+    _CONTEXT.set(Context())
+
+
+class _ContextualStream:
+    """A base class for streams that are contextually aware.
+
+    This class implements the TextIO interface via __getattr__ to delegate
+    attribute access to the original or contextual stream.
+    """
+    _original_stream: TextIO
+
+    def __init__(self, original_stream: TextIO):
+        self._original_stream = original_stream
+
+    def __getattr__(self, attr: str):
+        return getattr(self._active_stream(), attr)
+
+    def _active_stream(self) -> TextIO:
+        ctx = get()
+        if ctx is None:
+            return self._original_stream
+        return ctx.output_stream(self._original_stream)
+
+
+class Stdout(_ContextualStream):
+
+    def __init__(self):
+        super().__init__(sys.stdout)
+
+
+class Stderr(_ContextualStream):
+
+    def __init__(self):
+        super().__init__(sys.stderr)
diff --git a/sky/utils/context_utils.py b/sky/utils/context_utils.py
new file mode 100644
index 000000000..842b86ce1
--- /dev/null
+++ b/sky/utils/context_utils.py
@@ -0,0 +1,172 @@
+"""Utilities for SkyPilot context."""
+import asyncio
+import functools
+import io
+import multiprocessing
+import os
+import subprocess
+import sys
+import typing
+from typing import Any, Callable, IO, Optional, Tuple, TypeVar
+
+from sky import sky_logging
+from sky.utils import context
+from sky.utils import subprocess_utils
+
+StreamHandler = Callable[[IO[Any], IO[Any]], str]
+
+
+# TODO(aylei): call hijack_sys_attrs() proactivly in module init at server-side
+# once we have context widely adopted.
+def hijack_sys_attrs():
+    """hijack system attributes to be context aware
+
+    This function should be called at the very beginning of the processes
+    that might use sky.utils.context.
+    """
+    # Modify stdout and stderr of unvicorn process to be contextually aware,
+    # use setattr to bypass the TextIO type check.
+    setattr(sys, 'stdout', context.Stdout())
+    setattr(sys, 'stderr', context.Stderr())
+    # Reload logger to apply latest stdout and stderr.
+    sky_logging.reload_logger()
+    # Hijack os.environ with ContextualEnviron to make env variables
+    # contextually aware.
+    setattr(os, 'environ', context.ContextualEnviron(os.environ))
+    # Hijack subprocess.Popen to pass the contextual environ to subprocess
+    # by default.
+    setattr(subprocess, 'Popen', context.Popen)
+
+
+def passthrough_stream_handler(in_stream: IO[Any], out_stream: IO[Any]) -> str:
+    """Passthrough the stream from the process to the output stream"""
+    wrapped = io.TextIOWrapper(in_stream,
+                               encoding='utf-8',
+                               newline='',
+                               errors='replace',
+                               write_through=True)
+    while True:
+        line = wrapped.readline()
+        if line:
+            out_stream.write(line)
+            out_stream.flush()
+        else:
+            break
+    return ''
+
+
+def pipe_and_wait_process(
+        ctx: context.Context,
+        proc: subprocess.Popen,
+        poll_interval: float = 0.5,
+        cancel_callback: Optional[Callable[[], None]] = None,
+        stdout_stream_handler: Optional[StreamHandler] = None,
+        stderr_stream_handler: Optional[StreamHandler] = None
+) -> Tuple[str, str]:
+    """Wait for the process to finish or cancel it if the context is cancelled.
+
+    Args:
+        proc: The process to wait for.
+        poll_interval: The interval to poll the process.
+        cancel_callback: The callback to call if the context is cancelled.
+        stdout_stream_handler: An optional handler to handle the stdout stream,
+            if None, the stdout stream will be passed through.
+        stderr_stream_handler: An optional handler to handle the stderr stream,
+            if None, the stderr stream will be passed through.
+    """
+
+    if stdout_stream_handler is None:
+        stdout_stream_handler = passthrough_stream_handler
+    if stderr_stream_handler is None:
+        stderr_stream_handler = passthrough_stream_handler
+
+    # Threads are lazily created, so no harm if stderr is None
+    with multiprocessing.pool.ThreadPool(processes=2) as pool:
+        # Context will be lost in the new thread, capture current output stream
+        # and pass it to the new thread directly.
+        stdout_fut = pool.apply_async(
+            stdout_stream_handler, (proc.stdout, ctx.output_stream(sys.stdout)))
+        stderr_fut = None
+        if proc.stderr is not None:
+            stderr_fut = pool.apply_async(
+                stderr_stream_handler,
+                (proc.stderr, ctx.output_stream(sys.stderr)))
+        try:
+            wait_process(ctx,
+                         proc,
+                         poll_interval=poll_interval,
+                         cancel_callback=cancel_callback)
+        finally:
+            # Wait for the stream handler threads to exit when process is done
+            # or cancelled
+            stdout_fut.wait()
+            if stderr_fut is not None:
+                stderr_fut.wait()
+        stdout = stdout_fut.get()
+        stderr = ''
+        if stderr_fut is not None:
+            stderr = stderr_fut.get()
+        return stdout, stderr
+
+
+def wait_process(ctx: context.Context,
+                 proc: subprocess.Popen,
+                 poll_interval: float = 0.5,
+                 cancel_callback: Optional[Callable[[], None]] = None):
+    """Wait for the process to finish or cancel it if the context is cancelled.
+
+    Args:
+        proc: The process to wait for.
+        poll_interval: The interval to poll the process.
+        cancel_callback: The callback to call if the context is cancelled.
+    """
+    while True:
+        if ctx.is_canceled():
+            if cancel_callback is not None:
+                cancel_callback()
+            # Kill the process despite the caller's callback, the utility
+            # function gracefully handles the case where the process is
+            # already terminated.
+            subprocess_utils.kill_process_with_grace_period(proc)
+            raise asyncio.CancelledError()
+        try:
+            proc.wait(poll_interval)
+        except subprocess.TimeoutExpired:
+            pass
+        else:
+            # Process exited
+            break
+
+
+F = TypeVar('F', bound=Callable[..., Any])
+
+
+def cancellation_guard(func: F) -> F:
+    """Decorator to make a synchronous function cancellable via context.
+
+    Guards the function execution by checking context.is_canceled() before
+    executing the function and raises asyncio.CancelledError if the context
+    is already cancelled.
+
+    This basically mimics the behavior of asyncio, which checks coroutine
+    cancelled in await call.
+
+    Args:
+        func: The function to be decorated.
+
+    Returns:
+        The wrapped function that checks cancellation before execution.
+
+    Raises:
+        asyncio.CancelledError: If the context is cancelled before execution.
+    """
+
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        ctx = context.get()
+        if ctx is not None and ctx.is_canceled():
+            raise asyncio.CancelledError(
+                f'Function {func.__name__} cancelled before execution')
+        return func(*args, **kwargs)
+
+    return typing.cast(F, wrapper)
diff --git a/sky/utils/controller_utils.py b/sky/utils/controller_utils.py
index cd1788e4f..50db197ae 100644
--- a/sky/utils/controller_utils.py
+++ b/sky/utils/controller_utils.py
@@ -6,7 +6,7 @@ import getpass
 import os
 import tempfile
 import typing
-from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
+from typing import Any, Dict, Iterable, List, Optional, Set
 import uuid
 
 import colorama
@@ -517,6 +517,30 @@ def get_controller_resources(
         if custom_controller_resources_config is not None:
             controller_resources_config_copied.update(
                 custom_controller_resources_config)
+        # Compatibility with the old way of specifying the controller autostop
+        # config. TODO(cooperc): Remove this before 0.12.0.
+        custom_controller_autostop_config = skypilot_config.get_nested(
+            (controller.value.controller_type, 'controller', 'autostop'), None)
+        if custom_controller_autostop_config is not None:
+            logger.warning(
+                f'{colorama.Fore.YELLOW}Warning: Config value '
+                f'`{controller.value.controller_type}.controller.autostop` '
+                'is deprecated. Please use '
+                f'`{controller.value.controller_type}.controller.resources.'
+                f'autostop` instead.{colorama.Style.RESET_ALL}')
+            # Only set the autostop config if it is not already specified.
+            if controller_resources_config_copied.get('autostop') is None:
+                controller_resources_config_copied['autostop'] = (
+                    custom_controller_autostop_config)
+            else:
+                logger.warning(f'{colorama.Fore.YELLOW}Ignoring the old '
+                               'config, since it is already specified in '
+                               f'resources.{colorama.Style.RESET_ALL}')
+    # Set the default autostop config for the controller, if not already
+    # specified.
+    if controller_resources_config_copied.get('autostop') is None:
+        controller_resources_config_copied['autostop'] = (
+            controller.value.default_autostop_config)
 
     try:
         controller_resources = resources.Resources.from_yaml_config(
@@ -547,7 +571,10 @@ def get_controller_resources(
     if controller_record is not None:
         handle = controller_record.get('handle', None)
         if handle is not None:
-            controller_resources_to_use = handle.launched_resources
+            # Use the existing resources, but override the autostop config with
+            # the one currently specified in the config.
+            controller_resources_to_use = handle.launched_resources.copy(
+                autostop=controller_resources_config_copied.get('autostop'))
 
     # If the controller and replicas are from the same cloud (and region/zone),
     # it should provide better connectivity. We will let the controller choose
@@ -608,8 +635,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 +646,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:
@@ -641,40 +671,6 @@ def get_controller_resources(
     return result
 
 
-def get_controller_autostop_config(
-        controller: Controllers) -> Tuple[Optional[int], bool]:
-    """Get the autostop config for the controller.
-
-    Returns:
-      A tuple of (idle_minutes_to_autostop, down), which correspond to the
-      values passed to execution.launch().
-    """
-    controller_autostop_config_copied: Dict[str, Any] = copy.copy(
-        controller.value.default_autostop_config)
-    if skypilot_config.loaded():
-        custom_controller_autostop_config = skypilot_config.get_nested(
-            (controller.value.controller_type, 'controller', 'autostop'), None)
-        if custom_controller_autostop_config is False:
-            # Disabled with `autostop: false` in config.
-            # To indicate autostop is disabled, we return None for
-            # idle_minutes_to_autostop.
-            return None, False
-        elif custom_controller_autostop_config is True:
-            # Enabled with default values. There is no change in behavior, but
-            # this is included by for completeness, since `False` is valid.
-            pass
-        elif custom_controller_autostop_config is not None:
-            # We have specific config values.
-            # Override the controller autostop config with the ones specified in
-            # the config.
-            assert isinstance(custom_controller_autostop_config, dict)
-            controller_autostop_config_copied.update(
-                custom_controller_autostop_config)
-
-    return (controller_autostop_config_copied['idle_minutes'],
-            controller_autostop_config_copied['down'])
-
-
 def _setup_proxy_command_on_controller(
         controller_launched_cloud: 'clouds.Cloud',
         user_config: Dict[str, Any]) -> config_utils.Config:
@@ -703,7 +699,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
diff --git a/sky/utils/rich_utils.py b/sky/utils/rich_utils.py
index 606df16aa..53c759e17 100644
--- a/sky/utils/rich_utils.py
+++ b/sky/utils/rich_utils.py
@@ -1,13 +1,15 @@
 """Rich status spinner utils."""
 import contextlib
+import contextvars
 import enum
 import logging
 import threading
 import typing
-from typing import Dict, Iterator, Optional, Tuple, Union
+from typing import Callable, Iterator, Optional, Tuple, Union
 
 from sky.adaptors import common as adaptors_common
 from sky.utils import annotations
+from sky.utils import context
 from sky.utils import message_utils
 from sky.utils import rich_console_utils
 
@@ -18,11 +20,31 @@ else:
     requests = adaptors_common.LazyImport('requests')
     rich_console = adaptors_common.LazyImport('rich.console')
 
-_statuses: Dict[str, Optional[Union['EncodedStatus',
-                                    'rich_console.Status']]] = {
-                                        'server': None,
-                                        'client': None,
-                                    }
+GeneralStatus = Union['rich_console.Status', 'EncodedStatus']
+
+_client_status: Optional[GeneralStatus] = None
+_server_status: contextvars.ContextVar[
+    Optional[GeneralStatus]] = contextvars.ContextVar('server_status',
+                                                      default=None)
+
+
+def _get_client_status() -> Optional[GeneralStatus]:
+    return _client_status
+
+
+def _get_server_status() -> Optional[GeneralStatus]:
+    return _server_status.get()
+
+
+def _set_client_status(status: Optional[GeneralStatus]):
+    global _client_status
+    _client_status = status
+
+
+def _set_server_status(status: Optional[GeneralStatus]):
+    _server_status.set(status)
+
+
 _status_nesting_level = 0
 
 _logging_lock = threading.RLock()
@@ -128,20 +150,22 @@ class _NoOpConsoleStatus:
 class _RevertibleStatus:
     """A wrapper for status that can revert to previous message after exit."""
 
-    def __init__(self, message: str, status_type: str):
+    def __init__(self, message: str, get_status_fn: Callable[[], GeneralStatus],
+                 set_status_fn: Callable[[Optional[GeneralStatus]], None]):
         self.previous_message = None
-        self.status_type = status_type
-        status = _statuses[status_type]
+        self.get_status_fn = get_status_fn
+        self.set_status_fn = set_status_fn
+        status = self.get_status_fn()
         if status is not None:
             self.previous_message = status.status
         self.message = message
 
     def __enter__(self):
         global _status_nesting_level
-        _statuses[self.status_type].update(self.message)
+        self.get_status_fn().update(self.message)
         _status_nesting_level += 1
-        _statuses[self.status_type].__enter__()
-        return _statuses[self.status_type]
+        self.get_status_fn().__enter__()
+        return self.get_status_fn()
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         # We use the same lock with the `safe_logger` to avoid the following 2
@@ -160,32 +184,48 @@ class _RevertibleStatus:
             _status_nesting_level -= 1
             if _status_nesting_level <= 0:
                 _status_nesting_level = 0
-                if _statuses[self.status_type] is not None:
-                    _statuses[self.status_type].__exit__(
-                        exc_type, exc_val, exc_tb)
-                    _statuses[self.status_type] = None
+                if self.get_status_fn() is not None:
+                    self.get_status_fn().__exit__(exc_type, exc_val, exc_tb)
+                    self.set_status_fn(None)
             else:
-                _statuses[self.status_type].update(self.previous_message)
+                self.get_status_fn().update(self.previous_message)
 
     def update(self, *args, **kwargs):
-        _statuses[self.status_type].update(*args, **kwargs)
+        self.get_status_fn().update(*args, **kwargs)
 
     def stop(self):
-        _statuses[self.status_type].stop()
+        self.get_status_fn().stop()
 
     def start(self):
-        _statuses[self.status_type].start()
+        self.get_status_fn().start()
+
+
+def _is_thread_safe() -> bool:
+    """Check if the current status context is thread-safe.
+
+    We are thread-safe if we are on the main thread or the server_status is
+    context-local, i.e. an async context has been initialized.
+    """
+    return (threading.current_thread() is threading.main_thread() or
+            context.get() is not None)
 
 
 def safe_status(msg: str) -> Union['rich_console.Status', _NoOpConsoleStatus]:
-    """A wrapper for multi-threaded console.status."""
+    """A wrapper for multi-threaded server-side console.status.
+
+    This function will encode rich status with control codes and output the
+    encoded string to stdout. Client-side decode control codes from server
+    output and update the rich status. This function is safe to be called in
+    async/multi-threaded context.
+
+    See also: :func:`client_status`, :class:`EncodedStatus`.
+    """
     from sky import sky_logging  # pylint: disable=import-outside-toplevel
-    if (annotations.is_on_api_server and
-            threading.current_thread() is threading.main_thread() and
+    if (annotations.is_on_api_server and _is_thread_safe() and
             not sky_logging.is_silent()):
-        if _statuses['server'] is None:
-            _statuses['server'] = EncodedStatus(msg)
-        return _RevertibleStatus(msg, 'server')
+        if _get_server_status() is None:
+            _set_server_status(EncodedStatus(msg))
+        return _RevertibleStatus(msg, _get_server_status, _set_server_status)
     return _NoOpConsoleStatus()
 
 
@@ -196,22 +236,26 @@ def stop_safe_status():
     stream logs from user program and do not want it to interfere with the
     spinner display.
     """
-    if (threading.current_thread() is threading.main_thread() and
-            _statuses['server'] is not None):
-        _statuses['server'].stop()
+    if _is_thread_safe():
+        return
+    server_status = _get_server_status()
+    if server_status is not None:
+        server_status.stop()
 
 
 def force_update_status(msg: str):
     """Update the status message even if sky_logging.is_silent() is true."""
-    if (threading.current_thread() is threading.main_thread() and
-            _statuses['server'] is not None):
-        _statuses['server'].update(msg)
+    if not _is_thread_safe():
+        return
+    server_status = _get_server_status()
+    if server_status is not None:
+        server_status.update(msg)
 
 
 @contextlib.contextmanager
 def safe_logger():
     with _logging_lock:
-        client_status_obj = _statuses['client']
+        client_status_obj = _get_client_status()
 
         client_status_live = (client_status_obj is not None and
                               client_status_obj._live.is_started)  # pylint: disable=protected-access
@@ -230,13 +274,13 @@ class RichSafeStreamHandler(logging.StreamHandler):
 
 
 def client_status(msg: str) -> Union['rich_console.Status', _NoOpConsoleStatus]:
-    """A wrapper for multi-threaded console.status."""
+    """A wrapper for multi-threaded client-side console.status."""
     from sky import sky_logging  # pylint: disable=import-outside-toplevel
     if (threading.current_thread() is threading.main_thread() and
             not sky_logging.is_silent()):
-        if _statuses['client'] is None:
-            _statuses['client'] = rich_console_utils.get_console().status(msg)
-        return _RevertibleStatus(msg, 'client')
+        if _get_client_status() is None:
+            _set_client_status(rich_console_utils.get_console().status(msg))
+        return _RevertibleStatus(msg, _get_client_status, _set_client_status)
     return _NoOpConsoleStatus()
 
 
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index 4ed551e08..479c23cb7 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -33,6 +33,37 @@ def _check_not_both_fields_present(field1: str, field2: str):
     }
 
 
+_AUTOSTOP_SCHEMA = {
+    'anyOf': [
+        {
+            # Use boolean to disable autostop completely, e.g.
+            #   autostop: false
+            'type': 'boolean',
+        },
+        {
+            # Shorthand to set idle_minutes by directly specifying, e.g.
+            #   autostop: 5
+            'type': 'integer',
+            'minimum': 0,
+        },
+        {
+            'type': 'object',
+            'required': [],
+            'additionalProperties': False,
+            'properties': {
+                'idle_minutes': {
+                    'type': 'integer',
+                    'minimum': 0,
+                },
+                'down': {
+                    'type': 'boolean',
+                },
+            },
+        },
+    ],
+}
+
+
 def _get_single_resources_schema():
     """Schema for a single resource in a resources list."""
     # To avoid circular imports, only import when needed.
@@ -165,6 +196,7 @@ def _get_single_resources_schema():
                     'type': 'null',
                 }]
             },
+            'autostop': _AUTOSTOP_SCHEMA,
             # The following fields are for internal use only. Should not be
             # specified in the task config.
             '_docker_login_config': {
@@ -725,29 +757,6 @@ def get_config_schema():
         if k != '$schema'
     }
     resources_schema['properties'].pop('ports')
-    autostop_schema = {
-        'anyOf': [
-            {
-                # Use boolean to disable autostop completely, e.g.
-                #   autostop: false
-                'type': 'boolean',
-            },
-            {
-                'type': 'object',
-                'required': [],
-                'additionalProperties': False,
-                'properties': {
-                    'idle_minutes': {
-                        'type': 'integer',
-                        'minimum': 0,
-                    },
-                    'down': {
-                        'type': 'boolean',
-                    },
-                },
-            },
-        ],
-    }
     controller_resources_schema = {
         'type': 'object',
         'required': [],
@@ -762,7 +771,7 @@ def get_config_schema():
                     'high_availability': {
                         'type': 'boolean',
                     },
-                    'autostop': autostop_schema,
+                    'autostop': _AUTOSTOP_SCHEMA,
                 }
             },
             'bucket': {
diff --git a/sky/utils/subprocess_utils.py b/sky/utils/subprocess_utils.py
index 992b5d591..554d7cc65 100644
--- a/sky/utils/subprocess_utils.py
+++ b/sky/utils/subprocess_utils.py
@@ -208,8 +208,11 @@ def kill_children_processes(parent_pids: Optional[Union[
             kill_process_with_grace_period(child, force=force)
 
 
-def kill_process_with_grace_period(proc: Union[multiprocessing.Process,
-                                               psutil.Process],
+GenericProcess = Union[multiprocessing.Process, psutil.Process,
+                       subprocess.Popen]
+
+
+def kill_process_with_grace_period(proc: GenericProcess,
                                    force: bool = False,
                                    grace_period: int = 10) -> None:
     """Kill a process with SIGTERM and wait for it to exit.
@@ -223,6 +226,9 @@ def kill_process_with_grace_period(proc: Union[multiprocessing.Process,
     if isinstance(proc, psutil.Process):
         alive = proc.is_running
         wait = proc.wait
+    elif isinstance(proc, subprocess.Popen):
+        alive = lambda: proc.poll() is None
+        wait = proc.wait
     else:
         alive = proc.is_alive
         wait = proc.join


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}'
diff --git a/tests/unit_tests/test_controller_utils.py b/tests/unit_tests/test_controller_utils.py
index 28bfc6e3f..d3704afcd 100644
--- a/tests/unit_tests/test_controller_utils.py
+++ b/tests/unit_tests/test_controller_utils.py
@@ -8,32 +8,41 @@ from sky.jobs import constants as managed_job_constants
 from sky.serve import constants as serve_constants
 from sky.utils import controller_utils
 
+_DEFAULT_AUTOSTOP = {
+    'down': False,
+    'idle_minutes': 10,
+}
+
 
 @pytest.mark.parametrize(
     ('controller_type', 'custom_controller_resources_config', 'expected'), [
         ('jobs', {}, {
             'cpus': '4+',
             'memory': '8x',
-            'disk_size': 50
+            'disk_size': 50,
+            'autostop': _DEFAULT_AUTOSTOP,
         }),
         ('jobs', {
             'cpus': '4+',
-            'disk_size': 100
+            'disk_size': 100,
         }, {
             'cpus': '4+',
             'memory': '8x',
-            'disk_size': 100
+            'disk_size': 100,
+            'autostop': _DEFAULT_AUTOSTOP,
         }),
         ('serve', {}, {
             'cpus': '4+',
-            'disk_size': 200
+            'disk_size': 200,
+            'autostop': _DEFAULT_AUTOSTOP,
         }),
         ('serve', {
             'memory': '32+',
         }, {
             'cpus': '4+',
             'memory': '32+',
-            'disk_size': 200
+            'disk_size': 200,
+            'autostop': _DEFAULT_AUTOSTOP,
         }),
     ])
 def test_get_controller_resources(controller_type: str,
@@ -82,8 +91,14 @@ def _check_controller_resources(
 
 
 @pytest.mark.parametrize(('controller_type', 'default_controller_resources'), [
-    ('jobs', managed_job_constants.CONTROLLER_RESOURCES),
-    ('serve', serve_constants.CONTROLLER_RESOURCES),
+    ('jobs', {
+        **managed_job_constants.CONTROLLER_RESOURCES,
+        'autostop': _DEFAULT_AUTOSTOP,
+    }),
+    ('serve', {
+        **serve_constants.CONTROLLER_RESOURCES,
+        'autostop': _DEFAULT_AUTOSTOP,
+    }),
 ])
 def test_get_controller_resources_with_task_resources(
         controller_type: str, default_controller_resources: Dict[str, Any],
diff --git a/tests/unit_tests/test_sky/server/test_server.py b/tests/unit_tests/test_sky/server/test_server.py
index ec2110944..db681b4a8 100644
--- a/tests/unit_tests/test_sky/server/test_server.py
+++ b/tests/unit_tests/test_sky/server/test_server.py
@@ -2,6 +2,8 @@
 
 import argparse
 import asyncio
+import threading
+import time
 from unittest import mock
 
 import fastapi
@@ -10,6 +12,7 @@ import uvicorn
 
 from sky.server import server
 from sky.utils import common_utils
+from sky.utils import context
 
 
 @mock.patch('uvicorn.run')
@@ -135,7 +138,6 @@ async def test_validate():
 
     def slow_validate(*args, **kwargs):
         # Simulate slow validation
-        import time
         time.sleep(0.1)
         validation_complete.set()
 
@@ -155,3 +157,57 @@ async def test_validate():
         # Wait for validation to complete
         await validation_task
         assert validation_complete.is_set()
+
+
+@pytest.mark.asyncio
+async def test_logs():
+    """Test the logs endpoint."""
+    mock_cluster_job_body = mock.MagicMock()
+    mock_cluster_job_body.cluster_name = 'test-cluster'
+    mock_background_tasks = mock.MagicMock()
+
+    # Create an event to track when logs streaming starts
+    streaming_started = threading.Event()
+
+    # Mock the stream_response function
+    def mock_stream_response(*args, **kwargs):
+        streaming_started.set()
+        return fastapi.responses.StreamingResponse(
+            content=iter([]),  # Empty iterator for testing
+            media_type='text/plain')
+
+    def slow_execute(*args, **kwargs):
+        # Simulate slow execution
+        time.sleep(1)
+
+    with mock.patch('sky.server.requests.executor.prepare_request') as mock_prepare, \
+         mock.patch('sky.server.requests.executor.execute_request_coroutine',
+                   side_effect=slow_execute) as mock_execute, \
+         mock.patch('sky.server.stream_utils.stream_response',
+                   side_effect=mock_stream_response) as mock_stream:
+
+        # Mock prepare_request to return a request task
+        mock_request_task = mock.MagicMock()
+        mock_request_task.log_path = '/tmp/test.log'
+        mock_prepare.return_value = mock_request_task
+
+        # Start logs endpoint in background
+        logs_task = asyncio.create_task(
+            server.logs(mock.MagicMock(), mock_cluster_job_body,
+                        mock_background_tasks))
+
+        # Execute should be run in background and does not block streaming start
+        streaming_started.wait(timeout=0.1)
+
+        # Verify the response was created
+        response = await logs_task
+        assert isinstance(response, fastapi.responses.StreamingResponse)
+        assert response.media_type == 'text/plain'
+
+        # Verify the executor calls
+        mock_prepare.assert_called_once()
+        mock_execute.assert_called_once_with(mock_request_task)
+        mock_stream.assert_called_once_with(
+            request_id=mock.ANY,
+            logs_path=mock_request_task.log_path,
+            background_tasks=mock_background_tasks)
diff --git a/tests/unit_tests/test_sky/test_sky_logging.py b/tests/unit_tests/test_sky/test_sky_logging.py
new file mode 100644
index 000000000..e18556f6f
--- /dev/null
+++ b/tests/unit_tests/test_sky/test_sky_logging.py
@@ -0,0 +1,123 @@
+"""Unit tests for sky.sky_logging module."""
+
+import io
+import logging
+from unittest import mock
+
+import pytest
+
+from sky import sky_logging
+from sky.utils import context
+from sky.utils import env_options
+
+
+@pytest.fixture
+def stream():
+    """Fixture to provide a string IO stream."""
+    return io.StringIO()
+
+
+@pytest.fixture
+def handler(stream):
+    """Fixture to provide an EnvAwareHandler instance."""
+    return sky_logging.EnvAwareHandler(stream=stream, level=logging.INFO)
+
+
+@pytest.fixture
+def sensitive_handler(stream):
+    """Fixture to provide a sensitive EnvAwareHandler instance."""
+    return sky_logging.EnvAwareHandler(stream=stream,
+                                       level=logging.INFO,
+                                       sensitive=True)
+
+
+@mock.patch('sky.utils.context.get', return_value=None)
+def test_handler_default_level(mock_get, handler, monkeypatch):
+    """Test handler's default log level behavior."""
+    monkeypatch.setattr(env_options.Options.SHOW_DEBUG_INFO, 'get',
+                        lambda: False)
+    assert handler.level == logging.INFO
+    handler.level = logging.DEBUG
+    assert handler.level == logging.DEBUG
+
+
+@mock.patch('sky.utils.context.get', return_value=context.Context())
+def test_handler_with_debug_env(mock_get, handler, monkeypatch):
+    """Test handler respects SHOW_DEBUG_INFO env var."""
+    # Mock the env var to enable debug
+    monkeypatch.setattr(env_options.Options.SHOW_DEBUG_INFO, 'get',
+                        lambda: True)
+    assert handler.level == logging.DEBUG
+
+    # Mock the env var to disable debug
+    monkeypatch.setattr(env_options.Options.SHOW_DEBUG_INFO, 'get',
+                        lambda: False)
+    assert handler.level == logging.INFO
+
+
+@mock.patch('sky.utils.context.get', return_value=None)
+def test_handler_without_context(mock_get, handler, monkeypatch):
+    """Test handler behavior when no context is available."""
+    # Even with debug env var set, should return original level
+    monkeypatch.setattr(env_options.Options.SHOW_DEBUG_INFO, 'get',
+                        lambda: True)
+    assert handler.level == logging.INFO
+
+
+@mock.patch('sky.utils.context.get', return_value=context.Context())
+def test_sensitive_handler(mock_get, sensitive_handler, monkeypatch):
+    """Test sensitive handler's log level behavior."""
+    # Test when sensitive logs are suppressed
+    monkeypatch.setattr(env_options.Options.SUPPRESS_SENSITIVE_LOG, 'get',
+                        lambda: True)
+    monkeypatch.setattr(env_options.Options.SHOW_DEBUG_INFO, 'get',
+                        lambda: True)
+    assert sensitive_handler.level == logging.INFO
+
+    # Test when sensitive logs are not suppressed
+    monkeypatch.setattr(env_options.Options.SUPPRESS_SENSITIVE_LOG, 'get',
+                        lambda: False)
+    assert sensitive_handler.level == logging.DEBUG
+
+
+@mock.patch('sky.utils.context.get', return_value=context.Context())
+def test_handler_invalid_level(mock_get, handler, monkeypatch):
+    """Test setting invalid log levels."""
+    monkeypatch.setattr(env_options.Options.SHOW_DEBUG_INFO, 'get',
+                        lambda: False)
+    with pytest.raises(ValueError):
+        handler.level = 'INVALID_LEVEL'
+
+    # Test valid level setting
+    handler.level = logging.WARNING
+    assert handler._level == logging.WARNING
+
+
+@mock.patch('sky.utils.context.get', return_value=None)
+def test_handler_with_context_override(mock_get, handler, monkeypatch):
+    """Test setting invalid log levels."""
+    assert handler.level == logging.INFO
+    ctx = context.Context()
+    ctx.override_envs({'SKYPILOT_DEBUG': '1'})
+    mock_get.return_value = ctx
+    assert handler.level == logging.DEBUG
+
+
+@mock.patch('sky.utils.context.get', return_value=None)
+def test_handler_output(mock_get, handler, stream, monkeypatch):
+    """Sanity check."""
+    monkeypatch.setattr(env_options.Options.SHOW_DEBUG_INFO, 'get',
+                        lambda: False)
+    logger = logging.getLogger('test')
+    logger.addHandler(handler)
+    logger.setLevel(logging.DEBUG)
+
+    logger.debug('Debug message')
+    assert stream.getvalue() == ''
+
+    logger.info('Info message')
+    assert 'Info message' in stream.getvalue()
+
+    handler.level = logging.DEBUG
+    logger.debug('Debug message 2')
+    assert 'Debug message 2' in stream.getvalue()
diff --git a/tests/unit_tests/test_sky/utils/test_context_utils.py b/tests/unit_tests/test_sky/utils/test_context_utils.py
new file mode 100644
index 000000000..027d317e1
--- /dev/null
+++ b/tests/unit_tests/test_sky/utils/test_context_utils.py
@@ -0,0 +1,22 @@
+"""Unit tests for sky.utils.context_utils module."""
+from typing import Optional, Union
+
+from sky.utils import context_utils
+
+
+@context_utils.cancellation_guard
+def original_function(arg1: int, arg2: str) -> Optional[Union[int, str]]:
+    return None
+
+
+def test_cancellation_guard_perserves_typecheck():
+    # Verify that the decorated function has the same signature
+    assert original_function.__name__ == 'original_function'
+    assert original_function.__annotations__ == {
+        'arg1': int,
+        'arg2': str,
+        'return': Optional[Union[int, str]]
+    }
+
+    # Verify that the decorated function can be called with the same signature
+    assert original_function(1, 'test') is None
diff --git a/tests/unit_tests/test_sky/utils/text_context.py b/tests/unit_tests/test_sky/utils/text_context.py
new file mode 100644
index 000000000..a8ba88058
--- /dev/null
+++ b/tests/unit_tests/test_sky/utils/text_context.py
@@ -0,0 +1,270 @@
+"""Unit tests for sky.utils.context module."""
+
+import asyncio
+import os
+import pathlib
+import tempfile
+
+import pytest
+
+from sky.utils import context
+
+
+@pytest.fixture
+def ctx():
+    """Fixture to provide a fresh context for each test."""
+    context.initialize()
+    yield context.get()
+    # Cleanup
+    if context.get()._log_file_handle is not None:
+        context.get()._log_file_handle.close()
+
+
+def test_context_initialization():
+    """Test context initialization."""
+    assert context.get() is None
+    context.initialize()
+    assert isinstance(context.get(), context.Context)
+
+
+def test_context_cancellation(ctx):
+    """Test context cancellation."""
+    assert not ctx.is_canceled()
+    ctx.cancel()
+    assert ctx.is_canceled()
+
+
+def test_log_redirection(ctx):
+    """Test log file redirection."""
+    with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
+        log_path = pathlib.Path(f.name)
+
+    try:
+        # Test setting log file
+        old_log = ctx.redirect_log(log_path)
+        assert old_log is None
+        assert ctx._log_file == log_path
+        assert ctx._log_file_handle is not None
+
+        # Test removing log file
+        old_log = ctx.redirect_log(None)
+        assert old_log == log_path
+        assert ctx._log_file is None
+        assert ctx._log_file_handle is None
+    finally:
+        # Cleanup
+        if log_path.exists():
+            log_path.unlink()
+
+
+def test_env_overrides(ctx):
+    """Test environment variable overrides."""
+    # Setup env overrides
+    ctx.override_envs({'TEST_VAR': 'overridden'})
+    assert 'TEST_VAR' in ctx.env_overrides
+    assert ctx.env_overrides['TEST_VAR'] == 'overridden'
+
+    # Test adding more overrides
+    ctx.override_envs({'TEST_VAR2': 'overridden2'})
+    assert 'TEST_VAR2' in ctx.env_overrides
+    assert ctx.env_overrides['TEST_VAR2'] == 'overridden2'
+    # Original override should still be present
+    assert ctx.env_overrides['TEST_VAR'] == 'overridden'
+
+
+def test_contextual_environ_getitem(ctx):
+    """Test ContextualEnviron __getitem__ functionality."""
+    # Create a ContextualEnviron with a simple dict as base environ
+    base_environ = {'BASE_VAR': 'base_value'}
+    env = context.ContextualEnviron(base_environ)
+
+    # Test getting a value from base environ
+    assert env['BASE_VAR'] == 'base_value'
+
+    # Test getting a value that doesn't exist should raise KeyError
+    with pytest.raises(KeyError):
+        _ = env['NON_EXISTENT']
+
+    # Test override with context
+    ctx.override_envs({'BASE_VAR': 'overridden_value'})
+    assert env['BASE_VAR'] == 'overridden_value'
+
+    # Test adding a new override
+    ctx.override_envs({'NEW_VAR': 'new_value'})
+    assert env['NEW_VAR'] == 'new_value'
+
+
+def test_contextual_environ_iter(ctx):
+    """Test ContextualEnviron __iter__ functionality."""
+    # Create a ContextualEnviron with a simple dict as base environ
+    base_environ = {'BASE_VAR1': 'value1', 'BASE_VAR2': 'value2'}
+    env = context.ContextualEnviron(base_environ)
+
+    # Without overrides, should iterate base environ
+    keys = set(env)
+    assert keys == {'BASE_VAR1', 'BASE_VAR2'}
+
+    # With overrides, should include both base and overridden vars
+    ctx.override_envs({'NEW_VAR': 'new_value', 'BASE_VAR1': 'override'})
+    keys = set(env)
+    assert keys == {'BASE_VAR1', 'BASE_VAR2', 'NEW_VAR'}
+
+
+def test_contextual_environ_set_del(ctx):
+    """Test ContextualEnviron set and delete operations."""
+    # Create a ContextualEnviron with a simple dict as base environ
+    base_environ = {'BASE_VAR': 'base_value'}
+    env = context.ContextualEnviron(base_environ)
+
+    # Set a new value in the base environ
+    env['NEW_VAR'] = 'direct_value'
+    assert base_environ['NEW_VAR'] == 'direct_value'
+
+    # Delete a value from base environ
+    del env['BASE_VAR']
+    assert 'BASE_VAR' not in base_environ
+
+    # Context overrides should take precedence over base
+    ctx.override_envs({'NEW_VAR': 'override_value'})
+    assert env['NEW_VAR'] == 'override_value'
+    assert base_environ['NEW_VAR'] == 'direct_value'
+
+
+def test_contextual_environ_methods(ctx):
+    """Test other ContextualEnviron methods."""
+    # Create a ContextualEnviron with a simple dict as base environ
+    base_environ = {'BASE_VAR': 'base_value'}
+    env = context.ContextualEnviron(base_environ)
+    ctx.override_envs({'BASE_VAR': 'overridden_value'})
+
+    copied = env.copy()
+    assert copied == {'BASE_VAR': 'overridden_value'}
+    assert copied is not base_environ
+
+    # Test setdefault
+    assert env.setdefault('BASE_VAR', 'default') == 'base_value'
+    assert env.setdefault('NEW_DEFAULT', 'default_value') == 'default_value'
+    assert base_environ['NEW_DEFAULT'] == 'default_value'
+
+
+def test_contextual_environ_subprocess(ctx):
+    """Test ContextualEnviron behavior with subprocesses."""
+    import subprocess
+    import sys
+
+    # Save original os.environ to restore it later
+    original_environ = os.environ
+
+    try:
+        # Create a ContextualEnviron and replace os.environ
+        base_environ = dict(os.environ)
+        contextual_env = context.ContextualEnviron(base_environ)
+        os.environ = contextual_env
+
+        # Set an environment variable override in the context
+        test_var_name = 'SKYPILOT_TEST_SUBPROCESS_VAR'
+        ctx.override_envs({test_var_name: 'initial_value'})
+
+        assert os.getenv(test_var_name) == 'initial_value'
+
+        print_envs = (f'import os; '
+                      f'import time; '
+                      f'print(os.getenv(\'{test_var_name}\', \'NOT_FOUND\')); '
+                      'time.sleep(1); '
+                      f'print(os.getenv(\'{test_var_name}\', \'NOT_FOUND\'))')
+
+        # Run subprocess and capture its output
+        proc = context.Popen([sys.executable, '-c', print_envs],
+                             stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE,
+                             text=True)
+        ctx.override_envs({test_var_name: 'updated_value'})
+        proc.wait()
+
+        output = proc.stdout.readlines()
+        assert len(output) == 2
+        # The subprocess should see the context-overridden env vars at spawn time
+        assert output[0].strip() == 'initial_value'
+        # The subprocess should not see the updated value after spawn
+        assert output[1].strip() == 'initial_value'
+
+        # Verify the environment variable is updated in current process
+        assert os.getenv(test_var_name) == 'updated_value'
+    finally:
+        # Restore original os.environ
+        os.environ = original_environ
+
+
+def test_output_stream(ctx):
+    """Test output stream selection."""
+    import sys
+
+    # Default should return fallback
+    assert ctx.output_stream(sys.stdout) == sys.stdout
+
+    # With log file should return log file handle
+    with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
+        log_path = pathlib.Path(f.name)
+    try:
+        ctx.redirect_log(log_path)
+        assert ctx.output_stream(sys.stdout) == ctx._log_file_handle
+    finally:
+        if log_path.exists():
+            log_path.unlink()
+
+
+def test_contextual_streams(ctx):
+    """Test contextual stdout/stderr."""
+    import sys
+    stdout = context.Stdout()
+    stderr = context.Stderr()
+
+    # Without log file should use original streams
+    assert stdout._active_stream() == sys.stdout
+    assert stderr._active_stream() == sys.stderr
+
+    # With log file should use log file handle
+    with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
+        log_path = pathlib.Path(f.name)
+    try:
+        ctx.redirect_log(log_path)
+        assert stdout._active_stream() == ctx._log_file_handle
+        assert stderr._active_stream() == ctx._log_file_handle
+    finally:
+        if log_path.exists():
+            log_path.unlink()
+
+
+@pytest.mark.asyncio
+async def test_async_context_isolation():
+    """Test context isolation between coroutines."""
+    context.initialize()
+    ctx1 = context.get()
+
+    async def other_coro():
+        context.initialize()
+        return context.get()
+
+    ctx2 = await other_coro()
+    assert ctx1 is not ctx2
+
+
+@pytest.mark.asyncio
+async def test_async_cancellation():
+    """Test context cancellation in async environment."""
+    context.initialize()
+    ctx = context.get()
+
+    cancel_called = False
+
+    async def check_cancel():
+        nonlocal cancel_called
+        while not ctx.is_canceled():
+            await asyncio.sleep(0.1)
+        cancel_called = True
+
+    task = asyncio.create_task(check_cancel())
+    await asyncio.sleep(0.2)
+    ctx.cancel()
+    await asyncio.wait_for(task, timeout=1.0)
+    assert cancel_called