Repository: skypilot-org/skypilot
Pull request #6245 (2025-07-11 23:16:11)

[Hyperbolic] fix: mark `CUSTOM_MULTI_NETWORK ` un-supported
<!-- Describe the changes in this PR -->

Fix prevent cleanup_custom_multi_network error during Hyperbolic cluster teardown

**Problem**
`sky down` on Hyperbolic clusters was calling `cleanup_custom_multi_network()` and failing with `NotImplementedError` because Hyperbolic didn't declare this feature as unsupported.

**Solution**
- Added `CUSTOM_MULTI_NETWORK` to Hyperbolic's unsupported features list
- Implemented cleanup_custom_multi_network() function for proper routing
- Added necessary import in __init__.py

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [x] Code formatting: install pre-commit (auto-check on commit) or `bash format.sh`
- [x] Any manual or new tests for this PR (please specify below)
- [x] All smoke tests: `/smoke-test` (CI) or `pytest tests/test_smoke.py` (local)
- [x] Relevant individual tests: `/smoke-test -k test_name` (CI) or `pytest tests/test_smoke.py::test_name` (local)
- [x] 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/release-build.yml b/.github/workflows/release-build.yml
index f6749b2d6..74c7d3941 100644
--- a/.github/workflows/release-build.yml
+++ b/.github/workflows/release-build.yml
@@ -315,6 +315,7 @@ jobs:
 
   create-pr:
     needs: [release-build, smoke-tests, quicktest-core, release-tests]
+    if: always() && needs.release-build.result == 'success'
     runs-on: ubuntu-latest
     steps:
       - name: Checkout repository
diff --git a/.github/workflows/release-publish.yml b/.github/workflows/release-publish.yml
index cb2255427..03a38f9f1 100644
--- a/.github/workflows/release-publish.yml
+++ b/.github/workflows/release-publish.yml
@@ -114,6 +114,8 @@ jobs:
           fetch-depth: 0
 
       - name: Create and push tag
+        env:
+          GH_TOKEN: ${{ secrets.GH_PAT_FOR_RELEASE }}
         run: |
           VERSION="${{ needs.extract-version.outputs.version }}"
           git config --global user.name "GitHub Action"
@@ -128,6 +130,7 @@ jobs:
           tag_name: "v${{ needs.extract-version.outputs.version }}"
           generate_release_notes: true
           draft: true
+          token: ${{ secrets.GH_PAT_FOR_RELEASE }}
 
   cleanup-branches:
     needs: extract-version
diff --git a/docs/source/_static/custom.js b/docs/source/_static/custom.js
index f86673056..39628c51c 100644
--- a/docs/source/_static/custom.js
+++ b/docs/source/_static/custom.js
@@ -21,6 +21,7 @@ document.addEventListener('DOMContentLoaded', function () {
        r.src=t+h._hjSettings.hjid+j+h._hjSettings.hjsv;
        a.appendChild(r);
 })(window,document,'https://static.hotjar.com/c/hotjar-','.js?sv=');
+!function(t){var k="ko",i=(window.globalKoalaKey=window.globalKoalaKey||k);if(window[i])return;var ko=(window[i]=[]);["identify","track","removeListeners","on","off","qualify","ready"].forEach(function(t){ko[t]=function(){var n=[].slice.call(arguments);return n.unshift(t),ko.push(n),ko}});var n=document.createElement("script");n.async=!0,n.setAttribute("src","https://cdn.getkoala.com/v1/pk_d9bb6290ccb8a01b2d181fc0c8cf0dbb9836/sdk.js"),(document.body || document.head).appendChild(n)}();
 
 // New items: add 'new-item' class for for new items.
 document.addEventListener('DOMContentLoaded', () => {
diff --git a/docs/source/cloud-setup/cloud-permissions/gcp.rst b/docs/source/cloud-setup/cloud-permissions/gcp.rst
index f0a2bbb34..2d8fd4c1f 100644
--- a/docs/source/cloud-setup/cloud-permissions/gcp.rst
+++ b/docs/source/cloud-setup/cloud-permissions/gcp.rst
@@ -64,7 +64,7 @@ The easiest way to grant permissions to a user access your GCP project without t
   roles/compute.admin
   roles/iam.serviceAccountAdmin
   roles/iam.serviceAccountUser
-  roles/serviceusage.serviceUsageConsumer
+  roles/serviceusage.serviceUsageAdmin
   roles/storage.admin
   roles/iam.securityAdmin
 
diff --git a/docs/source/reference/config.rst b/docs/source/reference/config.rst
index a2eac6e9d..a8f3d661f 100644
--- a/docs/source/reference/config.rst
+++ b/docs/source/reference/config.rst
@@ -1281,10 +1281,11 @@ Example:
 .. code-block:: yaml
 
     nebius:
-        :ref:`use_internal_ips <config-yaml-nebius-use-internal-ips>`: true
-        :ref:`ssh_proxy_command <config-yaml-nebius-ssh-proxy-command>`: ssh -W %h:%p user@host
+        use_internal_ips: true
+        ssh_proxy_command:
           eu-north1: ssh -W %h:%p -p 1234 -o StrictHostKeyChecking=no myself@my.us-central1.proxy
           eu-west1: ssh -W %h:%p -i ~/.ssh/sky-key -o StrictHostKeyChecking=no nebiususer@<jump server public ip>
+        tenant_id: tenant-1234567890
         # Region-specific configuration
         eu-north1:
             # Project identifier for this region
@@ -1348,6 +1349,21 @@ Example:
       eu-north1: ssh -W %h:%p -p 1234 -o StrictHostKeyChecking=no myself@my.us-central1.proxy
       eu-west1: ssh -W %h:%p -i ~/.ssh/sky-key -o StrictHostKeyChecking=no nebiususer@<jump server public ip>
 
+.. _config-yaml-nebius-tenant-id:
+
+``nebius.tenant_id``
+~~~~~~~~~~~~~~~~~~~~
+
+Nebius tenant ID (optional).
+
+Example:
+
+.. code-block:: yaml
+
+  nebius:
+    tenant_id: tenant-1234567890
+
+
 .. _config-yaml-rbac:
 
 ``rbac``
diff --git a/sky/adaptors/nebius.py b/sky/adaptors/nebius.py
index 248991673..4f29e50c5 100644
--- a/sky/adaptors/nebius.py
+++ b/sky/adaptors/nebius.py
@@ -89,6 +89,12 @@ def iam():
     return iam_v1
 
 
+def billing():
+    # pylint: disable=import-outside-toplevel
+    from nebius.api.nebius.billing import v1alpha1 as billing_v1alpha1
+    return billing_v1alpha1
+
+
 def nebius_common():
     # pylint: disable=import-outside-toplevel
     from nebius.api.nebius.common import v1 as common_v1
diff --git a/sky/catalog/data_fetchers/fetch_nebius.py b/sky/catalog/data_fetchers/fetch_nebius.py
new file mode 100644
index 000000000..37f9cac24
--- /dev/null
+++ b/sky/catalog/data_fetchers/fetch_nebius.py
@@ -0,0 +1,308 @@
+"""A script that queries Nebius API to get instance types and pricing info.
+
+This script takes about 1 minute to finish.
+"""
+import csv
+from dataclasses import dataclass
+import decimal
+import json
+import logging
+import re
+from typing import Any, Dict, List, Optional
+
+from sky.adaptors import nebius
+from sky.adaptors.nebius import billing
+from sky.adaptors.nebius import compute
+from sky.adaptors.nebius import iam
+from sky.adaptors.nebius import nebius_common
+
+logger = logging.getLogger(__name__)
+
+TIMEOUT = 10
+PARENT_ID_TEMPLATE = 'project-{}public-images'
+ACCELERATOR_MANUFACTURER = 'NVIDIA'
+
+
+@dataclass
+class PresetInfo:
+    """Represents information about a specific compute preset,
+    including its pricing.
+
+    Attributes:
+        region (str): The geographical region where the preset is available.
+        fullname (str): The full name of the preset, a combination of platform
+        and preset name.
+        name (str): The name of the preset.
+        platform_name (str): The name of the platform the preset belongs to.
+        gpu (int): The number of GPUs in the preset.
+        vcpu (int): The number of virtual CPUs in the preset.
+        memory_gib (int): The amount of memory in GiB in the preset.
+        accelerator_manufacturer (str | None): The manufacturer of the
+        accelerator (e.g., "NVIDIA"), or None if no accelerator.
+        accelerator_name (str | None): The name of the accelerator
+        (e.g., "H100"), or None if no accelerator.
+        price_hourly (decimal.Decimal): The hourly price of the preset.
+    """
+
+    region: str
+    fullname: str
+    name: str
+    platform_name: str
+    gpu: int
+    vcpu: int
+    memory_gib: int
+    accelerator_manufacturer: Optional[str]
+    accelerator_name: Optional[str]
+    price_hourly: decimal.Decimal
+
+
+def _format_decimal(value: decimal.Decimal) -> str:
+    """Formats a decimal value to a string with at least two decimal places,
+    removing trailing zeros and ensuring a two-digit decimal part.
+
+    Args:
+        value (decimal.Decimal): The decimal value to format.
+
+    Returns:
+        str: The formatted string representation of the decimal.
+    """
+    formatted_value = f'{value:f}'.rstrip('0').rstrip('.')
+    integer_part, decimal_part = formatted_value.split(
+        '.') if '.' in formatted_value else (formatted_value, '')
+    if len(decimal_part) < 2:
+        decimal_part += '0' * (2 - len(decimal_part))
+
+    return f'{integer_part}.{decimal_part}'
+
+
+def _estimate_platforms(platforms: List[Any], parent_id: str,
+                        region: str) -> List[PresetInfo]:
+    """Collects specifications for all presets on the given platforms to form a
+    batch price request. It then sends the request and processes the responses
+    to create a list of PresetInfo objects.
+
+    Args:
+        platforms (List[Platform]): A List of compute platforms to estimate
+        prices for.
+        parent_id (str): The parent ID used for resource metadata
+        in the estimate request.
+        region (str): The region associated with the platforms.
+
+    Returns:
+        List[PresetInfo]: A list of PresetInfo objects containing details and
+        estimated prices for each preset.
+    """
+
+    calculator_service = billing().CalculatorServiceClient(nebius.sdk())
+    futures = []
+
+    for platform in platforms:
+        platform_name = platform.metadata.name
+
+        for preset in platform.spec.presets:
+            # Form the specification for the price request
+            estimate_spec = billing().ResourceSpec(
+                compute_instance_spec=compute().CreateInstanceRequest(
+                    metadata=nebius_common().ResourceMetadata(
+                        parent_id=parent_id,),
+                    spec=compute().InstanceSpec(
+                        resources=compute().ResourcesSpec(
+                            platform=platform_name,
+                            preset=preset.name,
+                        )),
+                ))
+
+            price_request = billing().EstimateBatchRequest(
+                resource_specs=[estimate_spec])
+            # Start future for each preset
+            futures.append((
+                platform,
+                preset,
+                calculator_service.estimate_batch(price_request,
+                                                  timeout=TIMEOUT),
+            ))
+
+    # wait all futures to complete and collect results
+    result = []
+    for platform, preset, future in futures:
+        platform_name = platform.metadata.name
+        result.append(
+            PresetInfo(
+                region=region,
+                fullname=f'{platform_name}_{preset.name}',
+                name=preset.name,
+                platform_name=platform_name,
+                gpu=preset.resources.gpu_count or 0,
+                vcpu=preset.resources.vcpu_count,
+                memory_gib=preset.resources.memory_gibibytes,
+                accelerator_manufacturer=ACCELERATOR_MANUFACTURER
+                if platform_name.startswith('gpu-') else '',
+                accelerator_name=platform_name.split('-')[1].upper()
+                if platform_name.startswith('gpu-') else '',
+                price_hourly=decimal.Decimal(
+                    future.wait().hourly_cost.general.total.cost),
+            ))
+
+    return result
+
+
+def _write_preset_prices(presets: List[PresetInfo], output_file: str) -> None:
+    """Writes the provided preset information to a CSV file.
+
+    Args:
+        presets (List[PresetInfo]): A list of PresetInfo objects to write.
+        output_file (str): The path to the output CSV file.
+    """
+    # Set up the CSV writer to output to stdout
+    with open(output_file, 'w', encoding='utf-8') as out:
+        header = [
+            'InstanceType',
+            'AcceleratorName',
+            'AcceleratorCount',
+            'vCPUs',
+            'MemoryGiB',
+            'Price',
+            'Region',
+            'GpuInfo',
+            'SpotPrice',
+        ]
+        writer = csv.DictWriter(out, fieldnames=header)
+        writer.writeheader()
+
+        for preset in sorted(presets,
+                             key=lambda x:
+                             (bool(x.gpu), x.region, x.platform_name, x.vcpu)):
+            gpu_info = ''
+            if preset.gpu > 0:
+                gpu_info_dict = {
+                    'Gpus': [{
+                        'Name': preset.accelerator_name,
+                        'Manufacturer': preset.accelerator_manufacturer,
+                        'Count': preset.gpu,
+                        'MemoryInfo': {
+                            'SizeInMiB': preset.memory_gib * 1024 // preset.gpu
+                        },
+                    }],
+                    'TotalGpuMemoryInMiB': preset.memory_gib * 1024,
+                }
+                gpu_info = json.dumps(gpu_info_dict).replace('"', '\'')
+
+            writer.writerow({
+                'InstanceType': preset.fullname,
+                'AcceleratorName': preset.accelerator_name,
+                'AcceleratorCount': preset.gpu,
+                'vCPUs': preset.vcpu,
+                'MemoryGiB': preset.memory_gib,
+                'Price': _format_decimal(preset.price_hourly),
+                'Region': preset.region,
+                'GpuInfo': gpu_info,
+                'SpotPrice': '',
+            })
+
+
+def _fetch_platforms_for_project(project_id: str) -> List[Any]:
+    """Fetches all available compute platforms for a given project.
+
+    Args:
+        project_id (str): The ID of the project to fetch platforms from.
+
+    Returns:
+        List[ComputePlatform]: A list of ComputePlatform objects available
+        in the project.
+    """
+    platform_service = compute().PlatformServiceClient(nebius.sdk())
+
+    platform_request = compute().ListPlatformsRequest(page_size=999,
+                                                      parent_id=project_id)
+    platform_response = platform_service.list(platform_request,
+                                              timeout=TIMEOUT).wait()
+
+    return platform_response.items
+
+
+def _get_regions_map() -> Dict[str, str]:
+    """Maps region codes to their full names by iterating through tenants and
+     projects.
+
+    Returns:
+        dict[str, str]: A dictionary where keys are region codes (e.g., "e00")
+                        and values are full region names (e.g., "eu-north1").
+    """
+    result = {}
+    response = iam().TenantServiceClient(nebius.sdk()).list(
+        iam().ListTenantsRequest(), timeout=TIMEOUT).wait()
+
+    for tenant in response.items:
+        projects = (iam().ProjectServiceClient(nebius.sdk()).list(
+            iam().ListProjectsRequest(parent_id=tenant.metadata.id),
+            timeout=TIMEOUT).wait())
+
+        for project in projects.items:
+            match = re.match(r'^project-([a-z0-9]{3})', project.metadata.id)
+            if match is None:
+                logger.error('Could not parse project id %s',
+                             project.metadata.id)
+                continue
+            result[match.group(1)] = project.status.region
+
+    return result
+
+
+def _get_all_platform_prices() -> List[PresetInfo]:
+    """Orchestrates fetching specifications and prices for all platforms across
+     all regions.
+
+    This function first retrieves a map of region codes to full names, then
+    iterates through each region, fetches available platforms for
+    the corresponding project ID, and finally estimates prices for all presets
+    on those platforms.
+
+    Returns:
+        List[PresetInfo]: A consolidated list of PresetInfo objects for all
+                        platforms and presets across all regions.
+    """
+
+    # Get regions codes to names
+    regions_map = _get_regions_map()
+
+    presets = []
+
+    for region_code in sorted(regions_map.keys()):
+        project_id = PARENT_ID_TEMPLATE.format(region_code)
+        region = regions_map[region_code]
+        logger.info('Processing region: %s (project: %s)...', region,
+                    project_id)
+
+        platforms = _fetch_platforms_for_project(project_id)
+        if not platforms:
+            logger.warning('No platforms found in region %s', region)
+            continue
+
+        presets.extend(
+            _estimate_platforms(platforms=platforms,
+                                parent_id=project_id,
+                                region=region))
+
+    return presets
+
+
+def main() -> None:
+    """Main function to fetch and write Nebius platform prices to a CSV file.
+
+    It initializes the SDK, fetches all platform prices, and then writes them
+    to the specified CSV file.
+    """
+
+    output_file = 'nebius/vms.csv'
+
+    # Fetch presets and estimate
+    presets = _get_all_platform_prices()
+
+    # Write CSV
+    _write_preset_prices(presets, output_file)
+
+    logger.info('Done!')
+
+
+if __name__ == '__main__':
+    main()
diff --git a/sky/client/cli/command.py b/sky/client/cli/command.py
index 745c2a19c..c8185bff0 100644
--- a/sky/client/cli/command.py
+++ b/sky/client/cli/command.py
@@ -151,13 +151,17 @@ def _get_cluster_records_and_set_ssh_config(
         if isinstance(handle.launched_resources.cloud, clouds.Kubernetes):
             # Replace the proxy command to proxy through the SkyPilot API
             # server with websocket.
-            key_path = (cluster_utils.SSHConfigHelper.generate_local_key_file(
-                handle.cluster_name, credentials))
+            escaped_key_path = shlex.quote(
+                (cluster_utils.SSHConfigHelper.generate_local_key_file(
+                    handle.cluster_name, credentials)))
+            escaped_executable_path = shlex.quote(sys.executable)
+            escaped_websocket_proxy_path = shlex.quote(
+                f'{sky.__root_dir__}/templates/websocket_proxy.py')
             # Instead of directly use websocket_proxy.py, we add an
             # additional proxy, so that ssh can use the head pod in the
             # cluster to jump to worker pods.
             proxy_command = (
-                f'ssh -tt -i {key_path} '
+                f'ssh -tt -i {escaped_key_path} '
                 '-o StrictHostKeyChecking=no '
                 '-o UserKnownHostsFile=/dev/null '
                 '-o IdentitiesOnly=yes '
@@ -167,10 +171,10 @@ def _get_cluster_records_and_set_ssh_config(
                 # TODO(zhwu): write the template to a temp file, don't use
                 # the one in skypilot repo, to avoid changing the file when
                 # updating skypilot.
-                f'\'{sys.executable} {sky.__root_dir__}/templates/'
-                f'websocket_proxy.py '
+                f'\"{escaped_executable_path} '
+                f'{escaped_websocket_proxy_path} '
                 f'{server_common.get_server_url()} '
-                f'{handle.cluster_name}\'')
+                f'{handle.cluster_name}\"')
             credentials['ssh_proxy_command'] = proxy_command
 
         cluster_utils.SSHConfigHelper.add_cluster(
diff --git a/sky/clouds/hyperbolic.py b/sky/clouds/hyperbolic.py
index d9cd2abac..5195bc99f 100644
--- a/sky/clouds/hyperbolic.py
+++ b/sky/clouds/hyperbolic.py
@@ -55,6 +55,8 @@ class Hyperbolic(clouds.Cloud):
             ('Auto-stop not supported.'),
         clouds.CloudImplementationFeatures.AUTODOWN:
             ('Auto-down not supported.'),
+        clouds.CloudImplementationFeatures.CUSTOM_MULTI_NETWORK:
+            ('Customized multiple network interfaces not supported.'),
     }
 
     PROVISIONER_VERSION = clouds.ProvisionerVersion.SKYPILOT
diff --git a/sky/clouds/nebius.py b/sky/clouds/nebius.py
index a7128c9ed..7535ae85f 100644
--- a/sky/clouds/nebius.py
+++ b/sky/clouds/nebius.py
@@ -219,10 +219,12 @@ class Nebius(clouds.Cloud):
             acc_dict)
         platform, _ = resources.instance_type.split('_')
 
-        if platform in ('cpu-d3', 'cpu-e2'):
-            image_family = 'ubuntu22.04-driverless'
-        elif platform in ('gpu-h100-sxm', 'gpu-h200-sxm', 'gpu-l40s-a'):
-            image_family = 'ubuntu22.04-cuda12'
+        # Selecting image_family by platform
+        # https://docs.nebius.com/compute/storage/boot-disk-images
+        if platform.startswith('cpu'):
+            image_family = 'ubuntu24.04-driverless'
+        elif platform.startswith('gpu'):
+            image_family = 'ubuntu24.04-cuda12'
         else:
             raise RuntimeError('Unsupported instance type for Nebius cloud:'
                                f' {resources.instance_type}')
diff --git a/sky/dashboard/src/components/clusters.jsx b/sky/dashboard/src/components/clusters.jsx
index ae424f606..c4e71acfc 100755
--- a/sky/dashboard/src/components/clusters.jsx
+++ b/sky/dashboard/src/components/clusters.jsx
@@ -80,14 +80,14 @@ const PROPERTY_OPTIONS = [
     label: 'User',
     value: 'user',
   },
-  {
-    label: 'Infra',
-    value: 'infra',
-  },
   {
     label: 'Workspace',
     value: 'workspace',
   },
+  {
+    label: 'Infra',
+    value: 'infra',
+  },
 ];
 
 // Helper function to format autostop information, similar to _get_autostop in CLI utils
@@ -354,7 +354,7 @@ export function Clusters() {
 
   return (
     <>
-      <div className="flex items-center justify-between mb-4 h-5">
+      <div className="flex items-center justify-between mb-2 h-5">
         <div className="text-base flex items-center">
           <Link
             href="/clusters"
@@ -363,7 +363,16 @@ export function Clusters() {
             Sky Clusters
           </Link>
 
-          <div className="flex items-center ml-6 space-x-3">
+          <FilterDropdown
+            propertyList={PROPERTY_OPTIONS}
+            valueList={optionValues}
+            setFilters={setFilters}
+            updateURLParams={updateURLParams}
+            placeholder="Filter clusters"
+          />
+        </div>
+        <div className="flex items-center space-x-4">
+          <div className="flex items-center">
             <label className="flex items-center cursor-pointer">
               <input
                 type="checkbox"
@@ -387,30 +396,22 @@ export function Clusters() {
               </span>
             </label>
           </div>
-
-          <FilterDropdown
-            propertyList={PROPERTY_OPTIONS}
-            valueList={optionValues}
-            setFilters={setFilters}
-            updateURLParams={updateURLParams}
-            placeholder="Filter clusters"
-          />
-        </div>
-        <div className="flex items-center">
-          {loading && (
-            <div className="flex items-center mr-2">
-              <CircularProgress size={15} className="mt-0" />
-              <span className="ml-2 text-gray-500">Loading...</span>
-            </div>
-          )}
-          <button
-            onClick={handleRefresh}
-            disabled={loading}
-            className="text-sky-blue hover:text-sky-blue-bright flex items-center"
-          >
-            <RotateCwIcon className="h-4 w-4 mr-1.5" />
-            {!isMobile && <span>Refresh</span>}
-          </button>
+          <div className="flex items-center">
+            {loading && (
+              <div className="flex items-center mr-2">
+                <CircularProgress size={15} className="mt-0" />
+                <span className="ml-2 text-gray-500">Loading...</span>
+              </div>
+            )}
+            <button
+              onClick={handleRefresh}
+              disabled={loading}
+              className="text-sky-blue hover:text-sky-blue-bright flex items-center"
+            >
+              <RotateCwIcon className="h-4 w-4 mr-1.5" />
+              {!isMobile && <span>Refresh</span>}
+            </button>
+          </div>
         </div>
       </div>
 
@@ -682,7 +683,7 @@ export function ClusterTable({
   return (
     <div>
       <Card>
-        <div className="overflow-x-auto">
+        <div className="overflow-x-auto rounded-lg">
           <Table className="min-w-full">
             <TableHeader>
               <TableRow>
@@ -1072,45 +1073,64 @@ const FilterDropdown = ({
   placeholder = 'Filter clusters',
 }) => {
   const inputRef = useRef(null);
+  const dropdownRef = useRef(null);
 
   const [isOpen, setIsOpen] = useState(false);
   const [value, setValue] = useState('');
   const [propertyValue, setPropertValue] = useState('status');
   const [valueOptions, setValueOptions] = useState([]);
 
-  let filteredOptions = [];
-
+  // Handle clicks outside the dropdown
   useEffect(() => {
-    let updatedValueOptions = [];
-
-    switch (propertyValue) {
-      case 'status':
-        updatedValueOptions = valueList.status;
-        break;
-
-      case 'user':
-        updatedValueOptions = valueList.user;
-        break;
-
-      case 'cluster':
-        updatedValueOptions = valueList.cluster;
-        break;
+    const handleClickOutside = (event) => {
+      if (
+        dropdownRef.current &&
+        !dropdownRef.current.contains(event.target) &&
+        inputRef.current &&
+        !inputRef.current.contains(event.target)
+      ) {
+        setIsOpen(false);
+      }
+    };
 
-      case 'workspace':
-        updatedValueOptions = valueList.workspace;
-        break;
+    document.addEventListener('mousedown', handleClickOutside);
+    return () => {
+      document.removeEventListener('mousedown', handleClickOutside);
+    };
+  }, []);
 
-      case 'infra':
-        updatedValueOptions = valueList.infra;
-        break;
+  useEffect(() => {
+    let updatedValueOptions = [];
 
-      default:
-        break;
+    if (valueList && typeof valueList === 'object') {
+      switch (propertyValue) {
+        case 'status':
+          updatedValueOptions = valueList.status || [];
+          break;
+        case 'user':
+          updatedValueOptions = valueList.user || [];
+          break;
+        case 'cluster':
+          updatedValueOptions = valueList.cluster || [];
+          break;
+        case 'workspace':
+          updatedValueOptions = valueList.workspace || [];
+          break;
+        case 'infra':
+          updatedValueOptions = valueList.infra || [];
+          break;
+        default:
+          break;
+      }
     }
 
-    updatedValueOptions = updatedValueOptions.filter((item) =>
-      item.includes(value.toLowerCase())
-    );
+    // Filter options based on current input value
+    if (value.trim() !== '') {
+      updatedValueOptions = updatedValueOptions.filter(
+        (item) =>
+          item && item.toString().toLowerCase().includes(value.toLowerCase())
+      );
+    }
 
     setValueOptions(updatedValueOptions);
   }, [propertyValue, valueList, value]);
@@ -1125,123 +1145,136 @@ const FilterDropdown = ({
 
   const handleValueChange = (e) => {
     setValue(e.target.value);
+    if (!isOpen) {
+      setIsOpen(true);
+    }
+  };
+
+  const handleInputFocus = () => {
+    setIsOpen(true);
+  };
+
+  const handleOptionSelect = (option) => {
+    setFilters((prevFilters) => {
+      const updatedFilters = [
+        ...prevFilters,
+        {
+          property: getPropertyLabel(propertyValue),
+          operator: ':',
+          value: option,
+        },
+      ];
+
+      updateURLParams(updatedFilters);
+      return updatedFilters;
+    });
+    setIsOpen(false);
+    setValue('');
+    inputRef.current.focus();
+  };
+
+  const handleKeyDown = (e) => {
+    if (e.key === 'Enter' && value.trim() !== '') {
+      setFilters((prevFilters) => {
+        const updatedFilters = [
+          ...prevFilters,
+          {
+            property: getPropertyLabel(propertyValue),
+            operator: ':',
+            value: value,
+          },
+        ];
+
+        updateURLParams(updatedFilters);
+        return updatedFilters;
+      });
+      setValue('');
+      setIsOpen(false);
+    } else if (e.key === 'Escape') {
+      setIsOpen(false);
+      inputRef.current.blur();
+    }
   };
 
   return (
-    <>
-      <div className="flex flex-row ml-4 mr-2 border border-gray-300 rounded-md">
-        <div className="border-r">
-          <Select
-            onValueChange={(value) => setPropertValue(value)}
-            value={propertyValue}
+    <div className="flex flex-row ml-4 mr-2 border border-gray-300 rounded-md overflow-visible">
+      <div className="border-r border-gray-300">
+        <Select onValueChange={setPropertValue} value={propertyValue}>
+          <SelectTrigger
+            aria-label="Filter Property"
+            className="focus:ring-0 focus:ring-offset-0 border-none rounded-l-md rounded-r-none w-32 h-8"
           >
-            <SelectTrigger
-              aria-label="Node"
-              className="focus:ring-0 focus:ring-offset-0 border-none w-36"
-            >
-              <SelectValue placeholder="Select Property" />
-            </SelectTrigger>
-            <SelectContent>
-              {propertyList.map((item, index) => (
-                <SelectItem key={`property-item-${index}`} value={item.value}>
-                  {item.label}
-                </SelectItem>
-              ))}
-            </SelectContent>
-          </Select>
-        </div>
-        <div className="relative">
-          <input
-            type="text"
-            ref={inputRef}
-            placeholder={placeholder}
-            value={value}
-            onChange={(e) => handleValueChange(e)}
-            onFocus={() => setIsOpen(true)}
-            onBlur={() => setIsOpen(false)}
-            onKeyDown={(e) => {
-              if (e.key === 'Enter' && value !== '') {
-                setFilters((prevFilters) => {
-                  const updatedFilters = [
-                    ...prevFilters,
-                    {
-                      property: getPropertyLabel(propertyValue),
-                      operator: ':',
-                      value: value,
-                    },
-                  ];
-
-                  updateURLParams(updatedFilters);
-
-                  return updatedFilters;
-                });
-                setValue('');
-                setIsOpen(false);
-                inputRef.current.blur();
-              }
+            <SelectValue placeholder="Select Property" />
+          </SelectTrigger>
+          <SelectContent>
+            {propertyList.map((item, index) => (
+              <SelectItem key={`property-item-${index}`} value={item.value}>
+                {item.label}
+              </SelectItem>
+            ))}
+          </SelectContent>
+        </Select>
+      </div>
+      <div className="relative flex-1">
+        <input
+          type="text"
+          ref={inputRef}
+          placeholder={placeholder}
+          value={value}
+          onChange={handleValueChange}
+          onFocus={handleInputFocus}
+          onKeyDown={handleKeyDown}
+          className="h-8 w-32 sm:w-96 px-3 pr-8 text-sm border-none rounded-l-none rounded-r-md focus:ring-0 focus:outline-none"
+          autoComplete="off"
+        />
+        {value && (
+          <button
+            onClick={() => {
+              setValue('');
+              setIsOpen(false);
             }}
-            className="h-10 w-32 sm:w-96 px-3 pr-8 text-sm rounded-md outline-none my-auto"
-            autoComplete="off"
-          />
-          {value && (
-            <button
-              onClick={() => {
-                setValue('');
-              }}
-              className="absolute right-2 top-1/2 transform -translate-y-1/2 text-gray-400 hover:text-gray-600"
-              title="Clear filter"
-              tabIndex={-1}
+            className="absolute right-2 top-1/2 transform -translate-y-1/2 text-gray-400 hover:text-gray-600"
+            title="Clear filter"
+            tabIndex={-1}
+          >
+            <svg
+              className="h-4 w-4"
+              fill="none"
+              stroke="currentColor"
+              viewBox="0 0 24 24"
             >
-              <svg
-                className="h-4 w-4"
-                fill="none"
-                stroke="currentColor"
-                viewBox="0 0 24 24"
+              <path
+                strokeLinecap="round"
+                strokeLinejoin="round"
+                strokeWidth={2}
+                d="M6 18L18 6M6 6l12 12"
+              />
+            </svg>
+          </button>
+        )}
+        {isOpen && valueOptions.length > 0 && (
+          <div
+            ref={dropdownRef}
+            className="absolute z-50 mt-1 w-full bg-white border border-gray-200 rounded-md shadow-lg max-h-60 overflow-y-auto"
+            style={{ zIndex: 9999 }}
+          >
+            {valueOptions.map((option, index) => (
+              <div
+                key={`${option}-${index}`}
+                className={`px-3 py-2 cursor-pointer hover:bg-gray-50 text-sm ${
+                  index !== valueOptions.length - 1
+                    ? 'border-b border-gray-100'
+                    : ''
+                }`}
+                onClick={() => handleOptionSelect(option)}
               >
-                <path
-                  strokeLinecap="round"
-                  strokeLinejoin="round"
-                  strokeWidth={2}
-                  d="M6 18L18 6M6 6l12 12"
-                />
-              </svg>
-            </button>
-          )}
-          {isOpen && valueOptions.length > 0 && (
-            <div className="flex flex-col absolute z-10 mt-1 w-full bg-white border border-gray-200 rounded-md shadow-lg">
-              {valueOptions.map((option, index) => (
-                <div
-                  key={option}
-                  className={`flex flex-col pl-7 py-2 cursor-pointer hover:bg-sky-50 text-sm ${index != filteredOptions.length - 1} && border-b`}
-                  onMouseDown={(e) => {
-                    e.preventDefault();
-                    setFilters((prevFilters) => {
-                      const updatedFilters = [
-                        ...prevFilters,
-                        {
-                          property: getPropertyLabel(propertyValue),
-                          operator: '=',
-                          value: option,
-                        },
-                      ];
-
-                      updateURLParams(updatedFilters);
-
-                      return updatedFilters;
-                    });
-                    setIsOpen(false);
-                    setValue('');
-                    inputRef.current.blur();
-                  }}
-                >
-                  <span className="text-sm text-gray-600">{option}</span>
-                </div>
-              ))}
-            </div>
-          )}
-        </div>
+                <span className="text-sm text-gray-700">{option}</span>
+              </div>
+            ))}
+          </div>
+        )}
       </div>
-    </>
+    </div>
   );
 };
 
@@ -1265,7 +1298,7 @@ const Filters = ({ filters = [], setFilters, updateURLParams }) => {
 
   return (
     <>
-      <div className="flex items-center gap-4 p-2">
+      <div className="flex items-center gap-4 py-2 px-2">
         <div className="flex flex-wrap items-content gap-2">
           {filters.map((filter, _index) => (
             <FilterItem
@@ -1279,7 +1312,7 @@ const Filters = ({ filters = [], setFilters, updateURLParams }) => {
             <>
               <button
                 onClick={clearFilters}
-                className="rounded-full px-4 py-1 text-gray-700 bg-gray-200 hover:bg-gray-300"
+                className="rounded-full px-4 py-1 text-sm text-gray-700 bg-gray-200 hover:bg-gray-300"
               >
                 Clear filters
               </button>
@@ -1294,16 +1327,16 @@ const Filters = ({ filters = [], setFilters, updateURLParams }) => {
 const FilterItem = ({ filter, onRemove }) => {
   return (
     <>
-      <div className="flex items-center gap-1 text-blue-600 bg-blue-100 px-2 rounded-full">
-        <div className="flex items-center gap-1 px-2 py-1 rounded-l-lg">
+      <div className="flex items-center text-blue-600 bg-blue-100 px-1 py-1 rounded-full text-sm">
+        <div className="flex items-center gap-1 px-2">
           <span>{`${filter.property} `}</span>
-          <span>{filter.operator}</span>
-          <span className="font-bold">{` ${filter.value}`}</span>
+          <span>{`${filter.operator} `}</span>
+          <span>{` ${filter.value}`}</span>
         </div>
 
         <button
           onClick={() => onRemove()}
-          className="p-1 transform text-gray-400 hover:text-gray-600 bg-blue-500 hover:bg-blue-600 rounded-full flex flex-col items-center"
+          className="p-0.5 ml-1 transform text-gray-400 hover:text-gray-600 bg-blue-500 hover:bg-blue-600 rounded-full flex flex-col items-center"
           title="Clear filter"
         >
           <svg
diff --git a/sky/dashboard/src/components/jobs.jsx b/sky/dashboard/src/components/jobs.jsx
index 0b44983d2..eeb8c4d61 100755
--- a/sky/dashboard/src/components/jobs.jsx
+++ b/sky/dashboard/src/components/jobs.jsx
@@ -853,7 +853,7 @@ export function ManagedJobsTable({
               <button
                 key={status}
                 onClick={() => handleStatusClick(status)}
-                className={`px-3 py-1 rounded-full flex items-center space-x-2 ${
+                className={`px-3 py-0.5 rounded-full flex items-center space-x-2 ${
                   isStatusHighlighted(status) ||
                   selectedStatuses.includes(status)
                     ? getBadgeStyle(status)
diff --git a/sky/execution.py b/sky/execution.py
index 80d9a4b15..d2c187c3d 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -23,6 +23,7 @@ from sky.utils import dag_utils
 from sky.utils import resources_utils
 from sky.utils import rich_utils
 from sky.utils import status_lib
+from sky.utils import tempstore
 from sky.utils import timeline
 from sky.utils import ux_utils
 
@@ -476,6 +477,9 @@ def _execute_dag(
 
 @timeline.event
 @usage_lib.entrypoint
+# A launch routine will share tempfiles between steps, so we init a tempdir
+# for the launch routine and gc the entire dir after launch.
+@tempstore.with_tempdir
 def launch(
     task: Union['sky.Task', 'sky.Dag'],
     cluster_name: Optional[str] = None,
diff --git a/sky/provision/hyperbolic/__init__.py b/sky/provision/hyperbolic/__init__.py
index cb1780b7a..df78d71e6 100644
--- a/sky/provision/hyperbolic/__init__.py
+++ b/sky/provision/hyperbolic/__init__.py
@@ -1,6 +1,7 @@
 """Hyperbolic provisioner for SkyPilot."""
 
 from sky.provision.hyperbolic.config import bootstrap_instances
+from sky.provision.hyperbolic.instance import cleanup_custom_multi_network
 from sky.provision.hyperbolic.instance import cleanup_ports
 from sky.provision.hyperbolic.instance import get_cluster_info
 from sky.provision.hyperbolic.instance import open_ports
diff --git a/sky/provision/hyperbolic/instance.py b/sky/provision/hyperbolic/instance.py
index 9e445c526..1be074dfe 100644
--- a/sky/provision/hyperbolic/instance.py
+++ b/sky/provision/hyperbolic/instance.py
@@ -414,6 +414,16 @@ def cleanup_ports(
     raise NotImplementedError('cleanup_ports is not supported for Hyperbolic')
 
 
+def cleanup_custom_multi_network(
+    cluster_name_on_cloud: str,
+    provider_config: Dict[str, Any],
+    failover: bool = False,
+) -> None:
+    """Cleanup custom multi-network. Not supported for Hyperbolic."""
+    raise NotImplementedError(
+        'cleanup_custom_multi_network is not supported for Hyperbolic')
+
+
 def open_ports(
     cluster_name_on_cloud: str,
     ports: list,
diff --git a/sky/resources.py b/sky/resources.py
index 69f3f6280..98a69c144 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -1730,6 +1730,8 @@ class Resources:
         if (blocked.accelerators is not None and
                 self.accelerators != blocked.accelerators):
             is_matched = False
+        if blocked.use_spot is not None and self.use_spot != blocked.use_spot:
+            is_matched = False
         return is_matched
 
     def is_empty(self) -> bool:
diff --git a/sky/server/requests/executor.py b/sky/server/requests/executor.py
index ace00920a..9a8bb1b93 100644
--- a/sky/server/requests/executor.py
+++ b/sky/server/requests/executor.py
@@ -176,7 +176,7 @@ class RequestWorker:
             if request_element is None:
                 time.sleep(0.1)
                 return
-            request_id, ignore_return_value, retryable = request_element
+            request_id, ignore_return_value, _ = request_element
             request = api_requests.get_request(request_id)
             assert request is not None, f'Request with ID {request_id} is None'
             if request.status == api_requests.RequestStatus.CANCELLED:
@@ -190,12 +190,10 @@ class RequestWorker:
             # the process, such as subprocess_daemon.py.
             fut = executor.submit_until_success(_request_execution_wrapper,
                                                 request_id, ignore_return_value)
-            if retryable:
-                # If the task might fail and be retried, start a thread to
-                # monitor the future and process retry.
-                threading.Thread(target=self.handle_task_result,
-                                 args=(fut, request_element),
-                                 daemon=True).start()
+            # Monitor the result of the request execution.
+            threading.Thread(target=self.handle_task_result,
+                             args=(fut, request_element),
+                             daemon=True).start()
 
             logger.info(f'[{self}] Submitted request: {request_id}')
         except (Exception, SystemExit) as e:  # pylint: disable=broad-except
@@ -209,6 +207,21 @@ class RequestWorker:
                            request_element: Tuple[str, bool, bool]) -> None:
         try:
             fut.result()
+        except concurrent.futures.process.BrokenProcessPool as e:
+            # Happens when the worker process dies unexpectedly, e.g. OOM
+            # killed.
+            request_id, _, retryable = request_element
+            # Ensure the request status.
+            api_requests.set_request_failed(request_id, e)
+            logger.error(
+                f'Request {request_id} failed to get processed '
+                f'{common_utils.format_exception(e, use_bracket=True)}')
+            if retryable:
+                # If the request is retryable and disrupted by broken
+                # process pool, reschedule it immediately to get it
+                # retried in the new process pool.
+                queue = _get_queue(self.schedule_type)
+                queue.put(request_element)
         except exceptions.ExecutionRetryableError as e:
             time.sleep(e.retry_wait_seconds)
             # Reschedule the request.
diff --git a/sky/server/requests/process.py b/sky/server/requests/process.py
index a8a17eb2e..1c0a74258 100644
--- a/sky/server/requests/process.py
+++ b/sky/server/requests/process.py
@@ -202,6 +202,9 @@ class BurstableExecutor:
                  burst_workers: int = 0,
                  **kwargs):
         if garanteed_workers > 0:
+            self._guaranteed_workers = garanteed_workers
+            self._guaranteed_pool_kwargs = kwargs
+            self._guaranteed_pool_lock = threading.Lock()
             self._executor = PoolExecutor(max_workers=garanteed_workers,
                                           **kwargs)
         if burst_workers > 0:
@@ -225,8 +228,7 @@ class BurstableExecutor:
 
         while True:
             if self._executor is not None and self._executor.has_idle_workers():
-                logger.info('Submitting to the guaranteed pool')
-                return self._executor.submit(fn, *args, **kwargs)
+                return self._submit_to_guaranteed_pool(fn, *args, **kwargs)
             if (self._burst_executor is not None and
                     self._burst_executor.has_idle_workers()):
                 try:
@@ -238,11 +240,52 @@ class BurstableExecutor:
             if self._executor is not None:
                 # No idle workers in either pool, still queue the request
                 # to the guaranteed pool to keep behavior consistent.
-                return self._executor.submit(fn, *args, **kwargs)
+                return self._submit_to_guaranteed_pool(fn, *args, **kwargs)
             logger.debug('No guaranteed pool set and the burst pool is full, '
                          'retry later.')
             time.sleep(0.1)
 
+    def _submit_to_guaranteed_pool(self, fn, *args,
+                                   **kwargs) -> concurrent.futures.Future:
+        """Submit to the guaranteed pool with retry.
+
+        The concurrent.futures.ProcessPoolExecutor will terminate all child
+        processes and reject new tasks if any of the child process died, e.g.
+        OOM killed. So when the pool is broken, we have to rebuild a new pool
+        to execute tasks.
+
+        Alternatives considered:
+        - multiprocessing.Pool: individual process failure does not affect the
+            pool, but it lacks of lazy-init support and does not handle process
+            failure gracefully.
+        - inherit from concurrent.futures.ProcessPoolExecutor: If a child
+            process dies when executing a task, we have to set an exception
+            in the Future object of the task. But there is a race condition
+            that the child process may dies exactly after get the task from
+            queue so there is no (obviously) safe way to record process PID
+            to task mapping in the current architecture of
+            ProcessPoolExecutor. It is more feasible to implement a custom
+            ProcessPoolExecutor with an new architecture.
+        - a custom ProcessPoolExecutor: non-trivial to implement, keep for
+            future improvement as BrokenProcessPool is not expected to be
+            a common case, i.e. it usually indicates there a bug elsewhere
+            when this happens.
+        """
+        with self._guaranteed_pool_lock:
+            assert self._executor is not None
+            try:
+                return self._executor.submit(fn, *args, **kwargs)
+            except concurrent.futures.process.BrokenProcessPool as e:
+                logger.warning('The guaranteed pool is broken, '
+                               f'replacing the pool and retrying. Error: {e}')
+                broken_pool = self._executor
+                threading.Thread(target=broken_pool.shutdown,
+                                 daemon=True).start()
+                self._executor = PoolExecutor(
+                    max_workers=self._guaranteed_workers,
+                    **self._guaranteed_pool_kwargs)
+        return self._submit_to_guaranteed_pool(fn, *args, **kwargs)
+
     def shutdown(self) -> None:
         """Shutdown the executor."""
 
diff --git a/sky/server/server.py b/sky/server/server.py
index 6294af62e..b0a1f8d4f 100644
--- a/sky/server/server.py
+++ b/sky/server/server.py
@@ -491,6 +491,9 @@ async def lifespan(app: fastapi.FastAPI):  # pylint: disable=redefined-outer-nam
                 func=event.run_event,
                 schedule_type=requests_lib.ScheduleType.SHORT,
                 is_skypilot_system=True,
+                # Request deamon should be retried if the process pool is
+                # broken.
+                retryable=True,
             )
         except exceptions.RequestAlreadyExistsError:
             # Lifespan will be executed in each uvicorn worker process, we
diff --git a/sky/setup_files/dependencies.py b/sky/setup_files/dependencies.py
index fdc642c3d..d2e128e86 100644
--- a/sky/setup_files/dependencies.py
+++ b/sky/setup_files/dependencies.py
@@ -129,6 +129,7 @@ extras_require: Dict[str, List[str]] = {
         'azure-mgmt-compute>=33.0.0',
         'azure-storage-blob>=12.23.1',
         'msgraph-sdk',
+        'msrestazure',
     ] + local_ray,
     # We need google-api-python-client>=2.69.0 to enable 'discardLocalSsd'
     # parameter for stopping instances. Reference:
@@ -173,7 +174,7 @@ extras_require: Dict[str, List[str]] = {
         # 'vsphere-automation-sdk @ git+https://github.com/vmware/vsphere-automation-sdk-python.git@v8.0.1.0' pylint: disable=line-too-long
     ],
     'nebius': [
-        'nebius>=0.2.0',
+        'nebius>=0.2.37',
     ] + aws_dependencies,
     'hyperbolic': [],  # No dependencies needed for hyperbolic
     'server': server_dependencies,
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index 70a80ea1e..9b9a90e36 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -956,8 +956,9 @@ class KubernetesCommandRunner(CommandRunner):
             return remote_home_dir
 
         # Build command.
-        helper_path = os.path.join(os.path.abspath(os.path.dirname(__file__)),
-                                   'kubernetes', 'rsync_helper.sh')
+        helper_path = shlex.quote(
+            os.path.join(os.path.abspath(os.path.dirname(__file__)),
+                         'kubernetes', 'rsync_helper.sh'))
         namespace_context = f'{self.namespace}+{self.context}'
         # Avoid rsync interpreting :, /, and + in namespace_context as the
         # default delimiter for options and arguments.
diff --git a/sky/utils/kubernetes/generate_kubeconfig.sh b/sky/utils/kubernetes/generate_kubeconfig.sh
index e15a7d3a8..1cbc8476e 100755
--- a/sky/utils/kubernetes/generate_kubeconfig.sh
+++ b/sky/utils/kubernetes/generate_kubeconfig.sh
@@ -156,6 +156,9 @@ rules:
   - apiGroups: [""]                 # Required for sky show-gpus command
     resources: ["pods"]
     verbs: ["get", "list"]
+  - apiGroups: ["rbac.authorization.k8s.io"]   # Required for SkyPilot to inspect its own permissions
+    resources: ["clusterroles", "clusterrolebindings", "roles", "rolebindings"]
+    verbs: ["get", "list", "watch"]
 ---
 # ClusterRoleBinding for the service account
 apiVersion: rbac.authorization.k8s.io/v1
diff --git a/sky/utils/tempstore.py b/sky/utils/tempstore.py
index b47be4926..06cce3a4c 100644
--- a/sky/utils/tempstore.py
+++ b/sky/utils/tempstore.py
@@ -2,9 +2,11 @@
 
 import contextlib
 import contextvars
+import functools
 import os
 import tempfile
-from typing import Iterator, Optional
+import typing
+from typing import Any, Callable, Iterator, Optional, TypeVar
 
 _TEMP_DIR: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
     'temp_store_dir', default=None)
@@ -49,3 +51,20 @@ def mkdtemp(suffix: Optional[str] = None,
         os.makedirs(dir, exist_ok=True)
 
     return tempfile.mkdtemp(suffix=suffix, prefix=prefix, dir=dir)
+
+
+F = TypeVar('F', bound=Callable[..., Any])
+
+
+def with_tempdir(func: F) -> F:
+    """Decorator that wraps a function call with tempdir() context manager.
+
+    Refer to `tempdir` for more details.
+    """
+
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        with tempdir():
+            return func(*args, **kwargs)
+
+    return typing.cast(F, wrapper)


Test patch:
diff --git a/tests/unit_tests/test_resources.py b/tests/unit_tests/test_resources.py
index ea13d7c2a..7a3f92d70 100644
--- a/tests/unit_tests/test_resources.py
+++ b/tests/unit_tests/test_resources.py
@@ -1047,3 +1047,92 @@ def test_priority_to_yaml_and_load(resources_kwargs, expected_yaml_config):
     assert loaded_r.cpus == r.cpus
     if 'accelerators' in expected_yaml_config:
         assert loaded_r.accelerators == r.accelerators
+
+
+@pytest.mark.parametrize(
+    'r_kwargs, blocked_kwargs, expected',
+    [
+        # All fields match
+        ({
+            'cloud': clouds.AWS(),
+            'instance_type': 'p3.2xlarge',
+            'region': 'us-west-2',
+            'zone': 'us-west-2a',
+            'accelerators': {
+                'V100': 1
+            },
+            'use_spot': True
+        }, {
+            'cloud': clouds.AWS(),
+            'instance_type': 'p3.2xlarge',
+            'region': 'us-west-2',
+            'zone': 'us-west-2a',
+            'accelerators': {
+                'V100': 1
+            },
+            'use_spot': True
+        }, True),
+        # use_spot mismatch
+        ({
+            'cloud': clouds.AWS(),
+            'instance_type': 'p3.2xlarge',
+            'region': 'us-west-2',
+            'zone': 'us-west-2a',
+            'accelerators': {
+                'V100': 1
+            },
+            'use_spot': True
+        }, {
+            'cloud': clouds.AWS(),
+            'instance_type': 'p3.2xlarge',
+            'region': 'us-west-2',
+            'zone': 'us-west-2a',
+            'accelerators': {
+                'V100': 1
+            },
+            'use_spot': False
+        }, False),
+        # cloud mismatch
+        ({
+            'cloud': clouds.AWS(),
+            'instance_type': 'p3.2xlarge',
+            'region': 'us-west-2',
+            'zone': 'us-west-2a',
+            'accelerators': {
+                'V100': 1
+            },
+        }, {
+            'cloud': clouds.GCP(),
+            'instance_type': 'g2-standard-4',
+            'region': 'us-east4',
+            'zone': 'us-east4-c',
+            'accelerators': {
+                'L4': 1
+            },
+        }, False),
+        # instance_type mismatch
+        ({
+            'cloud': clouds.AWS(),
+            'instance_type': 'p3.2xlarge',
+            'region': 'us-west-2',
+            'zone': 'us-west-2a',
+            'accelerators': {
+                'V100': 1
+            },
+            'use_spot': True
+        }, {
+            'cloud': clouds.AWS(),
+            'instance_type': 'p3.8xlarge',
+            'region': 'us-west-2',
+            'zone': 'us-west-2a',
+            'accelerators': {
+                'V100': 1
+            },
+            'use_spot': True
+        }, False),
+    ])
+def test_should_be_blocked_by(r_kwargs, blocked_kwargs, expected):
+    """Test should_be_blocked_by method."""
+    r = Resources(**r_kwargs)
+    blocked = Resources(**blocked_kwargs)
+    assert r.should_be_blocked_by(blocked) == expected
diff --git a/tests/unit_tests/test_sky/server/requests/test_process.py b/tests/unit_tests/test_sky/server/requests/test_process.py
index 494cc810d..591beefca 100644
--- a/tests/unit_tests/test_sky/server/requests/test_process.py
+++ b/tests/unit_tests/test_sky/server/requests/test_process.py
@@ -1,6 +1,8 @@
 """Unit tests for sky/server/requests/process.py."""
 from concurrent.futures import Future
+import concurrent.futures.process
 import time
+import unittest.mock
 
 import pytest
 
@@ -153,3 +155,50 @@ def test_burstable_executor_no_burst():
         executor.submit_until_success(dummy_task)
     finally:
         executor.shutdown()
+
+
+def test_burstable_executor_pool_recovery():
+    """Test BurstableExecutor recovery from BrokenProcessPool exception."""
+
+    executor = BurstableExecutor(garanteed_workers=1, burst_workers=0)
+
+    try:
+        # Store reference to original executor
+        original_executor = executor._executor
+        submit_call_count = 0
+
+        # Mock the PoolExecutor.submit method at class level to control
+        # behavior across all instances
+        original_submit = PoolExecutor.submit
+
+        def mock_submit(self, fn, *args, **kwargs):
+            nonlocal submit_call_count
+            submit_call_count += 1
+            if submit_call_count == 1:
+                # First call raises BrokenProcessPool to simulate pool failure
+                raise concurrent.futures.process.BrokenProcessPool(
+                    "Simulated process pool failure")
+            else:
+                # Subsequent calls should work normally
+                # Call the original submit method
+                return original_submit(self, fn, *args, **kwargs)
+
+        with unittest.mock.patch.object(PoolExecutor, 'submit',
+                                        new=mock_submit):
+            # This should trigger the pool recovery logic in
+            # _submit_to_guaranteed_pool
+            future = executor.submit_until_success(dummy_task)
+            result = future.result(timeout=5.0)
+
+            # Verify the task completed successfully despite initial failure
+            assert result is True
+
+            # Verify that submit was called at least twice
+            # (initial failure + successful retry)
+            assert submit_call_count >= 2
+
+            # Verify that a new executor was created after the failure
+            assert executor._executor is not original_executor
+
+    finally:
+        executor.shutdown()