Repository: skypilot-org/skypilot
Pull request #6173 (2025-07-04 05:54:05)

[Core] Cleanup temp dir for file_mounts after request
<!-- Describe the changes in this PR -->

We keep a tempdir for each launch request without GC, where the wheel file takes about ~1.5 MB and it can easily used up the local storage. This PR fixes the issue by cleanup the tempdir after request complete.

Tests:

- Unit tests
- Tested manually that the temp dir for file_mounts is deleted after launch

<!-- 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)
- [ ] 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 -->


aylei: /smoke-test -k basic

Patch:
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index a751869a5..226d29308 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -209,3 +209,150 @@ HOST=$(kubectl get svc ${RELEASE_NAME}-ingress-nginx-controller --namespace $NAM
 ENDPOINT=http://${WEB_USERNAME}:${WEB_PASSWORD}@${HOST}
 echo $ENDPOINT
 ```
+
+### Backward compatibility guidelines
+
+SkyPilot adopts a client-server achitecture and maintains backward compatibility between client and server to ensure smooth upgrades for users. Starting from `0.10.0`, SkyPilot guarantees compatibility between adjacent minor versions. That is:
+
+- Changes should always be added in a backward compatible manner, otherwise the compatibility for the previous minor version will be broken.
+- It is an opportunity to remove legacy compatibility code when a new minor version is released, e.g. when we release `0.12.0`, we can [remove the legacy compatibility code](#removing-compatibility-code) for `0.10.0` in our codebase.
+
+The general guideline to keep backward compatibility is to bump the `API_VERSION` constant in [`sky/server/constants.py`](https://github.com/skypilot-org/skypilot/blob/master/sky/server/constants.py) when introducing an API change and handle backward compatibility based on the `API_VERSION` of the remote peer:
+
+```diff
+# sky/server/constants.py
+- API_VERSION = 11
++ API_VERSION = 12
+
+# Application code
++ from sky.server import versions
++ # Handle the case where the remote peer runs in an API version older than 12
++ if versions.get_remote_api_version() < 12:
++   ...
+```
+
+Some concrete examples are listed below.
+
+#### Adding new APIs
+
+Bump the `API_VERSION` when adding a new API. Then:
+
+- For new SDK methods that calls the new API, add the `@versions.minimal_api_version(API_VERSION)` decorator to the method:
+
+    ```python
+    from sky.server import versions
+
+    # check_server_healthy_or_start is necessary to check and get server's API version,
+    # this decorator is typically added to all the SDK methods and will be omitted in the
+    # following examples.
+    @server_common.check_server_healthy_or_start
+    @versions.minimal_api_version(12)
+    def new_feature_method():
+        """This method requires server API version 12 or higher."""
+        pass
+    ```
+
+- For existing SDK methods that will be modified to call the new API, the business logic should handle backward compatibility based on the server's API version:
+
+    ```python
+    from sky.server import versions
+
+    def existing_sdk_method():
+        if versions.get_remote_api_version() >= 12:
+            # Call the new API
+        else:
+            # Proceed without the new API. Usually we just keep the same behavior
+            # as before the new API is introduced.
+    ```
+
+#### Adding new fields to API payload
+
+By convention, we define API payloads in `sky/server/api/payloads.py` and there are test cases to enforce the newly added fields must have a default value to keep backward compatibility at API level:
+
+- When receiving a payload from an older version without the new field, the default value is used for the missing new field.
+- When receiving a payload from a newer version with a new field, the value of the new field is ignored.
+
+However, when the value of the new field is taken from an user input (e.g. CLI flag), we should add a warning message to inform the user that the new field is ignored. An API version bump is required in this case. For example:
+
+```python
+from sky.server import versions
+
+@click.option('--newflag', default=None)
+def cli_entry_point(newflag: Optional[str] = None):
+    # The new flag is set but the server does not support the new field yet
+    if newflag is not None and versions.get_remote_api_version() < 12:
+        logger.warning('The new flag is ignored because the server does not support it yet.')
+```
+
+#### Refactoring existing APIs
+
+Refactoring existing APIs can be tricky. It is recommended to add an new API instead. Then the compatibility issue can be addressed in the same way as [Adding new APIs](#adding-new-apis), e.g.:
+
+- `constants.py`:
+
+    ```diff
+    - API_VERSION = 11
+    + API_VERSION = 12
+    ```
+
+- `server.py`:
+
+    ```python
+    @app.post('/api')
+    async def api():
+        ...
+
+    @app.post('/api_v2')
+    async def api_v2():
+        ...
+    ```
+
+- `sdk.py`:
+
+    ```python
+    from sky.server import versions
+
+    def sdk_method():
+        if versions.get_remote_api_version() >= 12:
+            # call /api_v2
+        else:
+            # call /api
+    ```
+
+If the refactoring is happen to be simple (e.g. just change the response payload structure), we can also directly modify the existing API. For example:
+
+- `server.py`:
+
+    ```python
+    from sky.server import versions
+
+    @app.post('/api')
+    async def api() -> Union[Response, ResponseV1]:
+        if versions.get_remote_api_version() >= 12:
+            return ResponseV2()
+        else:
+            return Response()
+    ```
+
+#### Removing compatibility code
+
+To reduce the maintenance burden, the SkyPilot CI pipeline automatically updates the `MIN_COMPATIBLE_API_VERSION` field in `sky/server/constants.py` when a new minor version is released.
+The SkyPilot CLI/SDK or API server will raise an error if the remote peer runs in an API version lower than `MIN_COMPATIBLE_API_VERSION`.
+Therefore, compatible code below this version can be safely removed in our codebase.
+
+For example, if we have added the example changes metioned in [Refactoring existing APIs](#refactoring-existing-apis) and now the `MIN_COMPATIBLE_API_VERSION` is bumped to 13 by the CI pipeline (which means we can drop compatibility for API version 12), then the following codes can be removed:
+
+```diff
+# sdk.py
+def sdk_method():
+- if versions.get_remote_api_version() >= 12:
+-     # call /api_v2
+- else:
+-     # call /api
++ # call /api_v2
+
+# server.py
+- @app.post('/api')
+- async def api():
+-    ...
+```
diff --git a/README.md b/README.md
index f7e506f1d..012b0318b 100644
--- a/README.md
+++ b/README.md
@@ -117,7 +117,8 @@ workdir: ~/torch_examples
 # Commands to be run before executing the job.
 # Typical use: pip install -r requirements.txt, git clone, etc.
 setup: |
-  pip install "torch<2.2" torchvision --index-url https://download.pytorch.org/whl/cu121
+  cd mnist
+  pip install -r requirements.txt
 
 # Commands to run as a job.
 # Typical use: launch the main program.
diff --git a/docs/source/conf.py b/docs/source/conf.py
index c988ed1b8..591890abd 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -181,7 +181,6 @@ exclude_patterns = [
     '_gallery_original',
     'generated-examples',
 ]
-myst_heading_anchors = 3
 myst_url_schemes = {
     'http': None,
     'https': None,
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 9e2d584c7..5987efcd0 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -51,6 +51,7 @@ from sky.utils import rich_utils
 from sky.utils import schemas
 from sky.utils import status_lib
 from sky.utils import subprocess_utils
+from sky.utils import tempstore
 from sky.utils import timeline
 from sky.utils import ux_utils
 from sky.workspaces import core as workspaces_core
@@ -247,7 +248,7 @@ def _optimize_file_mounts(tmp_yaml_path: str) -> None:
     #  - use a remote command to move all runtime files to their right places.
 
     # Local tmp dir holding runtime files.
-    local_runtime_files_dir = tempfile.mkdtemp()
+    local_runtime_files_dir = tempstore.mkdtemp()
     new_file_mounts = {_REMOTE_RUNTIME_FILES_DIR: local_runtime_files_dir}
 
     # Generate local_src -> unique_name.
diff --git a/sky/catalog/cudo_catalog.py b/sky/catalog/cudo_catalog.py
index adccccd0e..66cd8b73a 100644
--- a/sky/catalog/cudo_catalog.py
+++ b/sky/catalog/cudo_catalog.py
@@ -10,7 +10,7 @@ from sky.utils import ux_utils
 if typing.TYPE_CHECKING:
     from sky.clouds import cloud
 
-_PULL_FREQUENCY_HOURS = 1
+_PULL_FREQUENCY_HOURS = 7
 _df = common.read_catalog(cudo_mt.VMS_CSV,
                           pull_frequency_hours=_PULL_FREQUENCY_HOURS)
 
diff --git a/sky/client/sdk.py b/sky/client/sdk.py
index 70da4af0d..9410d1206 100644
--- a/sky/client/sdk.py
+++ b/sky/client/sdk.py
@@ -1710,11 +1710,11 @@ def get(request_id: str) -> Any:
     request_task = None
     if response.status_code == 200:
         request_task = requests_lib.Request.decode(
-            requests_lib.RequestPayload(**response.json()))
+            payloads.RequestPayload(**response.json()))
     elif response.status_code == 500:
         try:
             request_task = requests_lib.Request.decode(
-                requests_lib.RequestPayload(**response.json().get('detail')))
+                payloads.RequestPayload(**response.json().get('detail')))
             logger.debug(f'Got request with error: {request_task.name}')
         except Exception:  # pylint: disable=broad-except
             request_task = None
@@ -1855,7 +1855,7 @@ def api_status(
     request_ids: Optional[List[str]] = None,
     # pylint: disable=redefined-builtin
     all_status: bool = False
-) -> List[requests_lib.RequestPayload]:
+) -> List[payloads.RequestPayload]:
     """Lists all requests.
 
     Args:
@@ -1876,9 +1876,7 @@ def api_status(
         timeout=(client_common.API_SERVER_REQUEST_CONNECTION_TIMEOUT_SECONDS,
                  None))
     server_common.handle_request_error(response)
-    return [
-        requests_lib.RequestPayload(**request) for request in response.json()
-    ]
+    return [payloads.RequestPayload(**request) for request in response.json()]
 
 
 # === API server management APIs ===
diff --git a/sky/exceptions.py b/sky/exceptions.py
index 1fa28077a..eb8d7fb1f 100644
--- a/sky/exceptions.py
+++ b/sky/exceptions.py
@@ -509,6 +509,11 @@ class APIVersionMismatchError(RuntimeError):
     pass
 
 
+class APINotSupportedError(RuntimeError):
+    """Raised when the API is not supported by the remote peer."""
+    pass
+
+
 class JobExitCode(enum.IntEnum):
     """Job exit code enum.
 
diff --git a/sky/server/common.py b/sky/server/common.py
index 49a22121c..bc1989a37 100644
--- a/sky/server/common.py
+++ b/sky/server/common.py
@@ -22,7 +22,6 @@ import uuid
 import colorama
 import filelock
 
-import sky
 from sky import exceptions
 from sky import sky_logging
 from sky import skypilot_config
@@ -31,6 +30,7 @@ from sky.client import service_account_auth
 from sky.data import data_utils
 from sky.server import constants as server_constants
 from sky.server import rest
+from sky.server import versions
 from sky.skylet import constants
 from sky.usage import usage_lib
 from sky.utils import annotations
@@ -66,34 +66,11 @@ RETRY_COUNT_ON_TIMEOUT = 3
 # (e.g. in high contention env) and we will exit eagerly if server exit.
 WAIT_APISERVER_START_TIMEOUT_SEC = 60
 
-_VERSION_INFO = (
-    f'{colorama.Style.RESET_ALL}'
-    f'{colorama.Style.DIM}'
-    'client version: v{client_version} (API version: v{client_api_version})\n'
-    'server version: v{server_version} (API version: v{server_api_version})'
-    f'{colorama.Style.RESET_ALL}')
 _LOCAL_API_SERVER_RESTART_HINT = (
-    f'{colorama.Fore.YELLOW}Please restart the SkyPilot API server with:\n'
+    f'{colorama.Fore.YELLOW}The local SkyPilot API server is not compatible '
+    'with the client. Please restart the API server with:\n'
     f'{colorama.Style.BRIGHT}sky api stop; sky api start'
     f'{colorama.Style.RESET_ALL}')
-_LOCAL_SERVER_VERSION_MISMATCH_WARNING = (
-    f'{colorama.Fore.YELLOW}Client and local API server version mismatch:\n'
-    '{version_info}\n'
-    f'{_LOCAL_API_SERVER_RESTART_HINT}'
-    f'{colorama.Style.RESET_ALL}')
-_CLIENT_TOO_OLD_WARNING = (
-    f'{colorama.Fore.YELLOW}Your SkyPilot client is too old:\n'
-    '{version_info}\n'
-    f'{colorama.Fore.YELLOW}Upgrade your client with:\n'
-    '{command}'
-    f'{colorama.Style.RESET_ALL}')
-_REMOTE_SERVER_TOO_OLD_WARNING = (
-    f'{colorama.Fore.YELLOW}SkyPilot API server is too old:\n'
-    '{version_info}\n'
-    f'{colorama.Fore.YELLOW}Contact your administrator to upgrade the '
-    'remote API server or downgrade your local client with:\n'
-    '{command}\n'
-    f'{colorama.Style.RESET_ALL}')
 _SERVER_INSTALL_VERSION_MISMATCH_WARNING = (
     f'{colorama.Fore.YELLOW}SkyPilot API server version does not match the '
     'installation on disk:\n'
@@ -105,10 +82,6 @@ _SERVER_INSTALL_VERSION_MISMATCH_WARNING = (
     f'{colorama.Fore.YELLOW}This can happen if you upgraded SkyPilot without '
     'restarting the API server.'
     f'{colorama.Style.RESET_ALL}')
-# Parse local API version eargly to catch version format errors.
-_LOCAL_API_VERSION: int = int(server_constants.API_VERSION)
-# SkyPilot dev version.
-_DEV_VERSION = '1.0.0-dev0'
 
 RequestId = str
 ApiVersion = Optional[str]
@@ -134,6 +107,7 @@ class ApiServerInfo:
     commit: Optional[str] = None
     user: Optional[Dict[str, Any]] = None
     basic_auth_enabled: bool = False
+    error: Optional[str] = None
 
 
 def get_api_cookie_jar_path() -> pathlib.Path:
@@ -282,6 +256,23 @@ def is_api_server_local():
     return get_server_url() in AVAILABLE_LOCAL_API_SERVER_URLS
 
 
+def _handle_non_200_server_status(
+        response: 'requests.Response') -> ApiServerInfo:
+    if response.status_code == 401:
+        return ApiServerInfo(status=ApiServerStatus.NEEDS_AUTH)
+    if response.status_code == 400:
+        # Check if a version mismatch error is returned.
+        try:
+            body = response.json()
+            if (body.get('error',
+                         '') == ApiServerStatus.VERSION_MISMATCH.value):
+                return ApiServerInfo(status=ApiServerStatus.VERSION_MISMATCH,
+                                     error=body.get('message', ''))
+        except json.JSONDecodeError:
+            pass
+    return ApiServerInfo(status=ApiServerStatus.UNHEALTHY)
+
+
 def get_api_server_status(endpoint: Optional[str] = None) -> ApiServerInfo:
     """Retrieve the status of the API server.
 
@@ -315,10 +306,10 @@ def get_api_server_status(endpoint: Optional[str] = None) -> ApiServerInfo:
             return ApiServerInfo(status=ApiServerStatus.UNHEALTHY)
 
         logger.debug(f'Health check status: {response.status_code}')
-        if response.status_code == 401:
-            return ApiServerInfo(status=ApiServerStatus.NEEDS_AUTH)
-        elif response.status_code != 200:
-            return ApiServerInfo(status=ApiServerStatus.UNHEALTHY)
+
+        if response.status_code != 200:
+            return _handle_non_200_server_status(response)
+
         # The response is 200, so we can parse the response.
         try:
             result = response.json()
@@ -340,8 +331,24 @@ def get_api_server_status(endpoint: Optional[str] = None) -> ApiServerInfo:
                                f'version info. {server_url} may '
                                f'not be running SkyPilot API server.')
                 server_info.status = ApiServerStatus.UNHEALTHY
-            elif api_version != server_constants.API_VERSION:
-                server_info.status = ApiServerStatus.VERSION_MISMATCH
+            version_info = versions.check_compatibility_at_client(
+                response.headers)
+            if version_info is None:
+                # Backward compatibility for server prior to v0.11.0 which
+                # does not check compatibility at server side.
+                # TODO(aylei): remove this after v0.13.0 is released.
+                return ApiServerInfo(
+                    status=ApiServerStatus.VERSION_MISMATCH,
+                    error=versions.SERVER_TOO_OLD_ERROR.format(
+                        remote_version=version,
+                        local_version=versions.get_local_readable_version(),
+                        min_version=server_constants.MIN_COMPATIBLE_VERSION,
+                        command=versions.install_version_command(
+                            version, commit)))
+            if version_info.error is not None:
+                return ApiServerInfo(status=ApiServerStatus.VERSION_MISMATCH,
+                                     error=version_info.error)
+
             cookies = get_cookies_from_response(response)
             set_api_cookie_jar(cookies, create_if_not_exists=False)
             return server_info
@@ -490,7 +497,7 @@ def _start_api_server(deploy: bool = False,
         server_url = get_server_url(host)
         dashboard_msg = ''
         api_server_info = get_api_server_status(server_url)
-        if api_server_info.version == _DEV_VERSION:
+        if api_server_info.version == versions.DEV_VERSION:
             dashboard_msg += (
                 f'\n{colorama.Style.RESET_ALL}{ux_utils.INDENT_SYMBOL}'
                 f'{colorama.Fore.YELLOW}')
@@ -554,33 +561,11 @@ def check_server_healthy(
     api_server_info = get_api_server_status(endpoint)
     api_server_status = api_server_info.status
     if api_server_status == ApiServerStatus.VERSION_MISMATCH:
-        sv = api_server_info.api_version
-        assert sv is not None, 'Server API version is None'
-        try:
-            server_is_older = int(sv) < _LOCAL_API_VERSION
-        except ValueError:
-            # Raised when the server version using an unknown scheme.
-            # Version compatibility checking is expected to handle all legacy
-            # cases so we safely assume the server is newer when the version
-            # scheme is unknown.
-            logger.debug('API server version using unknown scheme: %s', sv)
-            server_is_older = False
-        version_info = _get_version_info_hint(api_server_info)
+        msg = api_server_info.error
         if is_api_server_local():
             # For local server, just hint user to restart the server to get
             # a consistent version.
-            msg = _LOCAL_SERVER_VERSION_MISMATCH_WARNING.format(
-                version_info=version_info)
-        else:
-            assert api_server_info.version is not None, 'Server version is None'
-            if server_is_older:
-                msg = _REMOTE_SERVER_TOO_OLD_WARNING.format(
-                    version_info=version_info,
-                    command=_install_server_version_command(api_server_info))
-            else:
-                msg = _CLIENT_TOO_OLD_WARNING.format(
-                    version_info=version_info,
-                    command=_install_server_version_command(api_server_info))
+            msg = _LOCAL_API_SERVER_RESTART_HINT
         with ux_utils.print_exception_no_traceback():
             raise exceptions.APIVersionMismatchError(msg)
     elif api_server_status == ApiServerStatus.UNHEALTHY:
@@ -614,37 +599,6 @@ def check_server_healthy(
     return api_server_status, api_server_info
 
 
-def _get_version_info_hint(server_info: ApiServerInfo) -> str:
-    assert server_info.version is not None, 'Server version is None'
-    # version_on_disk may be None if the server is older
-    assert server_info.commit is not None, 'Server commit is None'
-    sv = server_info.version
-    cv = sky.__version__
-    if server_info.version == _DEV_VERSION:
-        sv = f'{sv} with commit {server_info.commit}'
-    if cv == _DEV_VERSION:
-        cv = f'{cv} with commit {sky.__commit__}'
-    return _VERSION_INFO.format(client_version=cv,
-                                server_version=sv,
-                                client_api_version=server_constants.API_VERSION,
-                                server_api_version=server_info.api_version)
-
-
-def _install_server_version_command(server_info: ApiServerInfo) -> str:
-    assert server_info.version is not None, 'Server version is None'
-    assert server_info.commit is not None, 'Server commit is None'
-    if server_info.version == _DEV_VERSION:
-        # Dev build without valid version.
-        return ('pip install git+https://github.com/skypilot-org/skypilot@'
-                f'{server_info.commit}')
-    elif 'dev' in server_info.version:
-        # Nightly version.
-        return f'pip install -U "skypilot-nightly=={server_info.version}"'
-    else:
-        # Stable version.
-        return f'pip install -U "skypilot=={server_info.version}"'
-
-
 # Keep in sync with sky/setup_files/setup.py find_version()
 def get_skypilot_version_on_disk() -> str:
     """Get the version of the SkyPilot code on disk."""
diff --git a/sky/server/constants.py b/sky/server/constants.py
index 61db08fd6..5cdf77c5e 100644
--- a/sky/server/constants.py
+++ b/sky/server/constants.py
@@ -4,10 +4,31 @@ import os
 
 from sky.skylet import constants
 
-# API server version, whenever there is a change in API server that requires a
-# restart of the local API server or error out when the client does not match
-# the server version.
-API_VERSION = '10'
+# pylint: disable=line-too-long
+# The SkyPilot API version that the code currently use.
+# Bump this version when the API is changed and special compatibility handling
+# based on version info is needed.
+# For more details and code guidelines, refer to:
+# https://docs.skypilot.co/en/latest/developers/CONTRIBUTING.html#backward-compatibility-guidelines
+API_VERSION = 11
+
+# The minimum peer API version that the code should still work with.
+# Notes (dev):
+# - This value is maintained by the CI pipeline, DO NOT EDIT this manually.
+# - Compatibility code for versions lower than this can be safely removed.
+# Refer to API_VERSION for more details.
+MIN_COMPATIBLE_API_VERSION = 11
+
+# The semantic version of the minimum compatible API version.
+# Refer to MIN_COMPATIBLE_API_VERSION for more details.
+# Note (dev): DO NOT EDIT this constant manually.
+MIN_COMPATIBLE_VERSION = '0.10.0'
+
+# The HTTP header name for the API version of the sender.
+API_VERSION_HEADER = 'X-SkyPilot-API-Version'
+
+# The HTTP header name for the SkyPilot version of the sender.
+VERSION_HEADER = 'X-SkyPilot-Version'
 
 # Prefix for API request names.
 REQUEST_NAME_PREFIX = 'sky.'
diff --git a/sky/server/requests/executor.py b/sky/server/requests/executor.py
index 5bb40a342..1066dcc75 100644
--- a/sky/server/requests/executor.py
+++ b/sky/server/requests/executor.py
@@ -53,6 +53,7 @@ from sky.utils import common_utils
 from sky.utils import context
 from sky.utils import context_utils
 from sky.utils import subprocess_utils
+from sky.utils import tempstore
 from sky.utils import timeline
 from sky.workspaces import core as workspaces_core
 
@@ -338,6 +339,7 @@ def _request_execution_wrapper(request_id: str,
     2. Update the request status based on the execution result;
     3. Redirect the stdout and stderr of the execution to log file;
     4. Handle the SIGTERM signal to abort the request gracefully.
+    5. Maintain the lifecycle of the temp dir used by the request.
     """
     # Handle the SIGTERM signal to abort the request processing gracefully.
     signal.signal(signal.SIGTERM, _sigterm_handler)
@@ -361,7 +363,8 @@ def _request_execution_wrapper(request_id: str,
         # config, as there can be some logs during override that needs to be
         # captured in the log file.
         try:
-            with override_request_env_and_config(request_body):
+            with override_request_env_and_config(request_body), \
+                tempstore.tempdir():
                 if sky_logging.logging_enabled(logger, sky_logging.DEBUG):
                     config = skypilot_config.to_dict()
                     logger.debug(f'request config: \n'
diff --git a/sky/server/requests/payloads.py b/sky/server/requests/payloads.py
index bbb1dc24e..d7c1007a3 100644
--- a/sky/server/requests/payloads.py
+++ b/sky/server/requests/payloads.py
@@ -1,9 +1,27 @@
 """Payloads for the Sky API requests.
 
-TODO(zhwu): We can consider a better way to handle the default values of the
-kwargs for the payloads, otherwise, we have to keep the default values the sync
-with the backend functions. The benefit of having the default values in the
-payloads is that a user can find the default values in the Restful API docs.
+All the payloads that will be used between the client and server communication
+must be defined here to make sure it get covered by our API compatbility tests.
+
+Compatibility note:
+- Adding a new body for new API is compatible as long as the SDK method using
+  the new API is properly decorated with `versions.minimal_api_version`.
+- Adding a new field with default value to an existing body is compatible at
+  API level, but the business logic must handle the case where the field is
+  not proccessed by an old version of remote client/server. This can usually
+  be done by checking `versions.get_remote_api_version()`.
+- Other changes are not compatible at API level, so must be handled specially.
+  A common pattern is to keep both the old and new version of the body and
+  checking `versions.get_remote_api_version()` to decide which body to use. For
+  example, say we refactor the `LaunchBody`, the original `LaunchBody` must be
+  kept in the codebase and the new body should be added via `LaunchBodyV2`.
+  Then if the remote runs in an old version, the local code should still send
+  `LaunchBody` to keep the backward compatibility. `LaunchBody` can be removed
+  later when constants.MIN_COMPATIBLE_API_VERSION is updated to a version that
+  supports `LaunchBodyV2`
+
+Also refer to sky.server.constants.MIN_COMPATIBLE_API_VERSION and the
+sky.server.versions module for more details.
 """
 import os
 import typing
@@ -94,7 +112,18 @@ def get_override_skypilot_config_path_from_client() -> Optional[str]:
     return skypilot_config.loaded_config_path_serialized()
 
 
-class RequestBody(pydantic.BaseModel):
+class BasePayload(pydantic.BaseModel):
+    """The base payload for the SkyPilot API."""
+    # Ignore extra fields in the request body, which is useful for backward
+    # compatibility. The difference with `allow` is that `ignore` will not
+    # include the unknown fields when dump the model, i.e., we can add new
+    # fields to the request body without breaking the existing old API server
+    # where the handler function does not accept the new field in function
+    # signature.
+    model_config = pydantic.ConfigDict(extra='ignore')
+
+
+class RequestBody(BasePayload):
     """The request body for the SkyPilot API."""
     env_vars: Dict[str, str] = {}
     entrypoint: str = ''
@@ -103,11 +132,6 @@ class RequestBody(pydantic.BaseModel):
     override_skypilot_config: Optional[Dict[str, Any]] = {}
     override_skypilot_config_path: Optional[str] = None
 
-    # Allow extra fields in the request body, which is useful for backward
-    # compatibility, i.e., we can add new fields to the request body without
-    # breaking the existing old API server.
-    model_config = pydantic.ConfigDict(extra='allow')
-
     def __init__(self, **data):
         data['env_vars'] = data.get('env_vars', request_body_env_vars())
         usage_lib_entrypoint = usage_lib.messages.usage.entrypoint
@@ -665,3 +689,24 @@ class GetConfigBody(RequestBody):
 class CostReportBody(RequestBody):
     """The request body for the cost report endpoint."""
     days: Optional[int] = 30
+
+
+class RequestPayload(BasePayload):
+    """The payload for the requests."""
+
+    request_id: str
+    name: str
+    entrypoint: str
+    request_body: str
+    status: str
+    created_at: float
+    user_id: str
+    return_value: str
+    error: str
+    pid: Optional[int]
+    schedule_type: str
+    user_name: Optional[str] = None
+    # Resources the request operates on.
+    cluster_name: Optional[str] = None
+    status_msg: Optional[str] = None
+    should_retry: bool = False
diff --git a/sky/server/requests/requests.py b/sky/server/requests/requests.py
index 213586f1a..4b0d39008 100644
--- a/sky/server/requests/requests.py
+++ b/sky/server/requests/requests.py
@@ -98,28 +98,6 @@ class ScheduleType(enum.Enum):
     SHORT = 'short'
 
 
-@dataclasses.dataclass
-class RequestPayload:
-    """The payload for the requests."""
-
-    request_id: str
-    name: str
-    entrypoint: str
-    request_body: str
-    status: str
-    created_at: float
-    user_id: str
-    return_value: str
-    error: str
-    pid: Optional[int]
-    schedule_type: str
-    user_name: Optional[str] = None
-    # Resources the request operates on.
-    cluster_name: Optional[str] = None
-    status_msg: Optional[str] = None
-    should_retry: bool = False
-
-
 @dataclasses.dataclass
 class Request:
     """A SkyPilot API request."""
@@ -185,7 +163,7 @@ class Request:
     @classmethod
     def from_row(cls, row: Tuple[Any, ...]) -> 'Request':
         content = dict(zip(REQUEST_COLUMNS, row))
-        return cls.decode(RequestPayload(**content))
+        return cls.decode(payloads.RequestPayload(**content))
 
     def to_row(self) -> Tuple[Any, ...]:
         payload = self.encode()
@@ -194,7 +172,7 @@ class Request:
             row.append(getattr(payload, k))
         return tuple(row)
 
-    def readable_encode(self) -> RequestPayload:
+    def readable_encode(self) -> payloads.RequestPayload:
         """Serialize the SkyPilot API request for display purposes.
 
         This function should be called on the server side to serialize the
@@ -212,7 +190,7 @@ class Request:
                           payloads.RequestBody), (self.name, self.request_body)
         user = global_user_state.get_user(self.user_id)
         user_name = user.name if user is not None else None
-        return RequestPayload(
+        return payloads.RequestPayload(
             request_id=self.request_id,
             name=self.name,
             entrypoint=self.entrypoint.__name__,
@@ -230,12 +208,12 @@ class Request:
             should_retry=self.should_retry,
         )
 
-    def encode(self) -> RequestPayload:
+    def encode(self) -> payloads.RequestPayload:
         """Serialize the SkyPilot API request."""
         assert isinstance(self.request_body,
                           payloads.RequestBody), (self.name, self.request_body)
         try:
-            return RequestPayload(
+            return payloads.RequestPayload(
                 request_id=self.request_id,
                 name=self.name,
                 entrypoint=encoders.pickle_and_encode(self.entrypoint),
@@ -264,7 +242,7 @@ class Request:
             raise
 
     @classmethod
-    def decode(cls, payload: RequestPayload) -> 'Request':
+    def decode(cls, payload: payloads.RequestPayload) -> 'Request':
         """Deserialize the SkyPilot API request."""
         try:
             return cls(
diff --git a/sky/server/rest.py b/sky/server/rest.py
index 41112f490..d8a68bab9 100644
--- a/sky/server/rest.py
+++ b/sky/server/rest.py
@@ -12,6 +12,8 @@ import colorama
 from sky import exceptions
 from sky import sky_logging
 from sky.adaptors import common as adaptors_common
+from sky.server import constants
+from sky.server import versions
 from sky.utils import common_utils
 from sky.utils import rich_utils
 from sky.utils import ux_utils
@@ -28,6 +30,11 @@ F = TypeVar('F', bound=Callable[..., Any])
 
 _RETRY_CONTEXT = contextvars.ContextVar('retry_context', default=None)
 
+_session = requests.Session()
+_session.headers[constants.API_VERSION_HEADER] = str(constants.API_VERSION)
+_session.headers[constants.VERSION_HEADER] = (
+    versions.get_local_readable_version())
+
 
 class RetryContext:
 
@@ -132,13 +139,17 @@ def handle_server_unavailable(response: 'requests.Response') -> None:
 def request(method, url, **kwargs) -> 'requests.Response':
     """Send a request to the API server, retry on server temporarily
     unavailable."""
-    response = requests.request(method, url, **kwargs)
-    handle_server_unavailable(response)
-    return response
+    return request_without_retry(method, url, **kwargs)
 
 
 def request_without_retry(method, url, **kwargs) -> 'requests.Response':
     """Send a request to the API server without retry."""
-    response = requests.request(method, url, **kwargs)
+    response = _session.request(method, url, **kwargs)
     handle_server_unavailable(response)
+    remote_api_version = response.headers.get(constants.API_VERSION_HEADER)
+    remote_version = response.headers.get(constants.VERSION_HEADER)
+    if remote_api_version is not None:
+        versions.set_remote_api_version(int(remote_api_version))
+    if remote_version is not None:
+        versions.set_remote_version(remote_version)
     return response
diff --git a/sky/server/server.py b/sky/server/server.py
index 4b3eeecca..7ad04c979 100644
--- a/sky/server/server.py
+++ b/sky/server/server.py
@@ -4,7 +4,6 @@ import argparse
 import asyncio
 import base64
 import contextlib
-import dataclasses
 import datetime
 import hashlib
 import json
@@ -49,6 +48,7 @@ from sky.server import constants as server_constants
 from sky.server import metrics
 from sky.server import state
 from sky.server import stream_utils
+from sky.server import versions
 from sky.server.requests import executor
 from sky.server.requests import payloads
 from sky.server.requests import preconditions
@@ -559,6 +559,35 @@ class GracefulShutdownMiddleware(starlette.middleware.base.BaseHTTPMiddleware):
         return await call_next(request)
 
 
+class APIVersionMiddleware(starlette.middleware.base.BaseHTTPMiddleware):
+    """Middleware to add API version to the request."""
+
+    async def dispatch(self, request: fastapi.Request, call_next):
+        version_info = versions.check_compatibility_at_server(request.headers)
+        # Bypass version handling for backward compatibility with clients prior
+        # to v0.11.0, the client will check the version in the body of
+        # /api/health response and hint an upgrade.
+        # TODO(aylei): remove this after v0.13.0 is released.
+        if version_info is None:
+            return await call_next(request)
+        if version_info.error is None:
+            versions.set_remote_api_version(version_info.api_version)
+            versions.set_remote_version(version_info.version)
+            response = await call_next(request)
+        else:
+            response = fastapi.responses.JSONResponse(
+                status_code=400,
+                content={
+                    'error': common.ApiServerStatus.VERSION_MISMATCH.value,
+                    'message': version_info.error,
+                })
+        response.headers[server_constants.API_VERSION_HEADER] = str(
+            server_constants.API_VERSION)
+        response.headers[server_constants.VERSION_HEADER] = \
+            versions.get_local_readable_version()
+        return response
+
+
 app = fastapi.FastAPI(prefix='/api/v1', debug=True, lifespan=lifespan)
 # Middleware wraps in the order defined here. E.g., given
 #   app.add_middleware(Middleware1)
@@ -571,6 +600,8 @@ app = fastapi.FastAPI(prefix='/api/v1', debug=True, lifespan=lifespan)
 # Use environment variable to make the metrics middleware optional.
 if os.environ.get(constants.ENV_VAR_SERVER_METRICS_ENABLED):
     app.add_middleware(metrics.PrometheusMiddleware)
+app.add_middleware(APIVersionMiddleware)
+app.add_middleware(RBACMiddleware)
 app.add_middleware(InternalDashboardPrefixMiddleware)
 app.add_middleware(GracefulShutdownMiddleware)
 app.add_middleware(PathCleanMiddleware)
@@ -1335,7 +1366,7 @@ async def local_down(request: fastapi.Request) -> None:
 
 # === API server related APIs ===
 @app.get('/api/get')
-async def api_get(request_id: str) -> requests_lib.RequestPayload:
+async def api_get(request_id: str) -> payloads.RequestPayload:
     """Gets a request with a given request ID prefix."""
     while True:
         request_task = requests_lib.get_request(request_id)
@@ -1350,9 +1381,8 @@ async def api_get(request_id: str) -> requests_lib.RequestPayload:
                     detail=f'Request {request_id!r} should be retried')
             request_error = request_task.get_error()
             if request_error is not None:
-                raise fastapi.HTTPException(status_code=500,
-                                            detail=dataclasses.asdict(
-                                                request_task.encode()))
+                raise fastapi.HTTPException(
+                    status_code=500, detail=request_task.encode().model_dump())
             return request_task.encode()
         # yield control to allow other coroutines to run, sleep shortly
         # to avoid storming the DB and CPU in the meantime
@@ -1490,7 +1520,7 @@ async def api_status(
         None, description='Request IDs to get status for.'),
     all_status: bool = fastapi.Query(
         False, description='Get finished requests as well.'),
-) -> List[requests_lib.RequestPayload]:
+) -> List[payloads.RequestPayload]:
     """Gets the list of requests."""
     if request_ids is None:
         statuses = None
@@ -1530,7 +1560,10 @@ async def health(request: fastapi.Request) -> Dict[str, Any]:
     logger.info(f'Health endpoint: request.state.auth_user = {user}')
     return {
         'status': common.ApiServerStatus.HEALTHY.value,
-        'api_version': server_constants.API_VERSION,
+        # Kept for backward compatibility, clients before 0.11.0 will read this
+        # field to check compatibility and hint the user to upgrade the CLI.
+        # TODO(aylei): remove this field after 0.13.0
+        'api_version': str(server_constants.API_VERSION),
         'version': sky.__version__,
         'version_on_disk': common.get_skypilot_version_on_disk(),
         'commit': sky.__commit__,
diff --git a/sky/server/uvicorn.py b/sky/server/uvicorn.py
index 6415aa967..2161fa69f 100644
--- a/sky/server/uvicorn.py
+++ b/sky/server/uvicorn.py
@@ -159,6 +159,8 @@ class Server(uvicorn.Server):
     def run(self, *args, **kwargs):
         """Run the server process."""
         context_utils.hijack_sys_attrs()
+        # Use default loop policy of uvicorn (use uvloop if available).
+        self.config.setup_event_loop()
         with self.capture_signals():
             asyncio.run(self.serve(*args, **kwargs))
 
diff --git a/sky/server/versions.py b/sky/server/versions.py
new file mode 100644
index 000000000..c4d57d040
--- /dev/null
+++ b/sky/server/versions.py
@@ -0,0 +1,270 @@
+"""API versioning module."""
+
+import contextvars
+import functools
+import re
+from typing import Callable, Literal, Mapping, NamedTuple, Optional, Tuple
+
+import colorama
+from packaging import version as version_lib
+
+import sky
+from sky import exceptions
+from sky import sky_logging
+from sky.server import constants
+from sky.utils import ux_utils
+
+logger = sky_logging.init_logger(__name__)
+
+CLIENT_TOO_OLD_ERROR = (
+    f'{colorama.Fore.YELLOW}Your SkyPilot client version is too old:'
+    '{remote_version}\n'
+    f'{colorama.Style.RESET_ALL}'
+    'The server is running on {local_version} and the minimum compatible '
+    'version is {min_version}.\n'
+    f'Upgrade your client with:\n{colorama.Fore.YELLOW}'
+    '{command}'
+    f'{colorama.Style.RESET_ALL}')
+SERVER_TOO_OLD_ERROR = (
+    f'{colorama.Fore.YELLOW}Your SkyPilot API server version is too old: '
+    '{remote_version}\n'
+    f'{colorama.Style.RESET_ALL}'
+    'The client is running on {local_version} and the minimum compatible '
+    'version is {min_version}.\n'
+    'Contact your administrator to upgrade the remote API server or downgrade '
+    f'your client with:\n{colorama.Fore.YELLOW}'
+    '{command}'
+    f'{colorama.Style.RESET_ALL}')
+
+# SkyPilot dev version.
+DEV_VERSION = '1.0.0-dev0'
+
+_REMOTE_TO_ERROR = {
+    'client': CLIENT_TOO_OLD_ERROR,
+    'server': SERVER_TOO_OLD_ERROR,
+}
+
+# Context-local (thread or cooroutine) remote API version, captured during
+# communication with the remote peer.
+_remote_api_version: contextvars.ContextVar[Optional[int]] = \
+    contextvars.ContextVar('remote_api_version', default=None)
+_remote_version: contextvars.ContextVar[str] = \
+    contextvars.ContextVar('remote_version', default='unknown')
+_reminded_for_minor_version_upgrade = False
+
+
+def get_remote_api_version() -> Optional[int]:
+    return _remote_api_version.get()
+
+
+def set_remote_api_version(api_version: int) -> None:
+    _remote_api_version.set(api_version)
+
+
+def get_remote_version() -> str:
+    return _remote_version.get()
+
+
+def set_remote_version(version: str) -> None:
+    _remote_version.set(version)
+
+
+class VersionInfo(NamedTuple):
+    api_version: int
+    version: str
+    error: Optional[str] = None
+
+
+def check_compatibility_at_server(
+        client_headers: Mapping[str, str]) -> Optional[VersionInfo]:
+    """Check API compatibility between client and server."""
+    return _check_version_compatibility(client_headers, 'client')
+
+
+def check_compatibility_at_client(
+        server_headers: Mapping[str, str]) -> Optional[VersionInfo]:
+    """Check API compatibility between client and server."""
+    return _check_version_compatibility(server_headers, 'server')
+
+
+def _check_version_compatibility(
+        remote_headers: Mapping[str, str],
+        remote_type: Literal['client', 'server']) -> Optional[VersionInfo]:
+    """Check API compatibility between client and server.
+
+    This function can be called at both client and server side, where the
+    headers should contain the version info of the remote.
+
+    Args:
+        remote_headers: The headers of the request/response sent from the
+            remote.
+        remote_type: The type of the remote, used to determine the error
+            message. Valid options are 'client' and 'server'.
+
+    Returns:
+        The version info of the remote, None if the version info is not found
+        in the headers for backward compatibility.
+    """
+    api_version_str = remote_headers.get(constants.API_VERSION_HEADER)
+    version = remote_headers.get(constants.VERSION_HEADER)
+    if version is None or api_version_str is None:
+        return None
+    try:
+        api_version = int(api_version_str)
+    except ValueError:
+        # The future change is expected to not break the compatibility of this
+        # header, so we are encountering a bug or a malicious request here,
+        # just raise an error.
+        raise ValueError(
+            f'Header {constants.API_VERSION_HEADER}: '
+            f'{api_version_str} is not a valid API version.') from None
+
+    if api_version < constants.MIN_COMPATIBLE_API_VERSION:
+        if remote_type == 'server':
+            # Hint the user to downgrade to client to the remote server server.
+            server_version, server_commit = parse_readable_version(version)
+            command = install_version_command(server_version, server_commit)
+        else:
+            # Hint the client to upgrade to upgrade the server version
+            command = install_version_command(sky.__version__, sky.__commit__)
+        return VersionInfo(api_version=api_version,
+                           version=version,
+                           error=_REMOTE_TO_ERROR[remote_type].format(
+                               remote_version=version,
+                               local_version=get_local_readable_version(),
+                               min_version=constants.MIN_COMPATIBLE_VERSION,
+                               command=command,
+                           ))
+
+    if remote_type == 'server':
+        # Only print the reminder at client-side.
+        _remind_minor_version_upgrade(version)
+
+    return VersionInfo(api_version=api_version, version=version)
+
+
+def get_local_readable_version() -> str:
+    """Get the readable version of the SkyPilot code loaded in current process.
+
+    For dev version, the version is formatted as: 1.0.0-dev0 (commit: 1234567)
+    to make it meaningful for users.
+    """
+    if sky.__version__ == DEV_VERSION:
+        return f'{sky.__version__} (commit: {sky.__commit__})'
+    else:
+        return sky.__version__
+
+
+def parse_readable_version(version: str) -> Tuple[str, Optional[str]]:
+    """Parse a readable produced by get_local_readable_version.
+
+    Args:
+        version: The version string to parse.
+
+    Returns:
+        A tuple of (version, optional_commit) where:
+        - version: The base version string (e.g., "1.0.0-dev0")
+        - optional_commit: The commit hash if present, None otherwise
+    """
+    # Check if this is a dev version with commit info
+    # Format: "1.0.0-dev0 (commit: 1234567)"
+    commit_pattern = r'^(.+) \(commit: ([a-f0-9]+)\)$'
+    match = re.match(commit_pattern, version)
+
+    if match:
+        base_version = match.group(1)
+        commit = match.group(2)
+        return base_version, commit
+    else:
+        # Regular version without commit info
+        return version, None
+
+
+def install_version_command(version: str, commit: Optional[str] = None) -> str:
+    if version == DEV_VERSION:
+        if commit is not None:
+            return ('pip install git+https://github.com/skypilot-org/skypilot@'
+                    f'{commit}')
+    elif 'dev' in version:
+        return f'pip install -U "skypilot-nightly=={version}"'
+    return f'pip install -U "skypilot=={version}"'
+
+
+def _remind_minor_version_upgrade(remote_version: str) -> None:
+    """Remind the user to upgrade the CLI/SDK."""
+    # Only print the reminder once per process.
+    global _reminded_for_minor_version_upgrade
+    if _reminded_for_minor_version_upgrade:
+        return
+    # Skip for dev versions.
+    if 'dev' in sky.__version__ or 'dev' in remote_version:
+        return
+
+    # Remove the commit info if any.
+    remote_base_version, _ = parse_readable_version(remote_version)
+
+    # Parse semver for both local and remote versions
+    try:
+        local = version_lib.parse(sky.__version__)
+        remote = version_lib.parse(remote_base_version)
+
+        # Check if local version is behind remote version, ignore patch version.
+        if (local.major, local.minor) < (remote.major, remote.minor):
+            logger.warning(
+                f'{colorama.Fore.YELLOW}The SkyPilot API server is running in '
+                f'version {remote_version}, which is newer than your client '
+                f'version {sky.__version__}. The compatibility for your '
+                f'current version might be dropped in the next server upgrade.'
+                f'\nConsider upgrading your client with:\n'
+                f'{install_version_command(remote_version)}'
+                f'{colorama.Style.RESET_ALL}')
+            _reminded_for_minor_version_upgrade = True
+    except version_lib.InvalidVersion:
+        # Skip for non-valid semver (probabely a dev version)
+        pass
+
+
+# TODO(aylei): maybe we can use similiar approach to mark a new argument can
+# only be used in the new server version.
+def minimal_api_version(min_version: int) -> Callable:
+    """Decorator to enforce a minimum remote API version for an SDK function.
+
+    New SDK method must be decorated with this decorator to make sure it raises
+    an readable error when the remote server is not upgraded.
+
+    Args:
+        min_version: The minimum remote API version required to call the
+        function.
+
+    Returns:
+        A decorator function that checks API version before execution.
+
+    Raises:
+        APINotSupportedError: If the remote API version is below the minimum
+            required.
+    """
+
+    def decorator(func: Callable) -> Callable:
+
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            remote_api_version = get_remote_api_version()
+            if remote_api_version is None:
+                return func(*args, **kwargs)
+            if remote_api_version < min_version:
+                with ux_utils.print_exception_no_traceback():
+                    hint = 'Please upgrade the remote server.'
+                    # The client runs in a released version, do better hint.
+                    if 'dev' not in sky.__version__:
+                        hint = (
+                            f'Upgrade the remote server to {sky.__version__} '
+                            'and re-run the command.')
+                    raise exceptions.APINotSupportedError(
+                        f'Function {func.__name__} is introduced after the '
+                        f'remote server version {get_remote_version()!r} is '
+                        f'released. {hint}')
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator
diff --git a/sky/utils/tempstore.py b/sky/utils/tempstore.py
new file mode 100644
index 000000000..b47be4926
--- /dev/null
+++ b/sky/utils/tempstore.py
@@ -0,0 +1,51 @@
+"""Temporary storage context manager."""
+
+import contextlib
+import contextvars
+import os
+import tempfile
+from typing import Iterator, Optional
+
+_TEMP_DIR: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
+    'temp_store_dir', default=None)
+
+
+@contextlib.contextmanager
+def tempdir() -> Iterator[str]:
+    """Context manager for temporary directory of current context.
+
+    This wraps tempfile.TemporaryDirectory and makes the temp dir available
+    throughout the context, eliminating the need to pass the temp dir to
+    the nested functions that need it.
+
+    This context manager is nestable - nested calls will create new temp dirs
+    and restore the previous temp dir when exiting.
+    """
+    with tempfile.TemporaryDirectory(prefix='sky-tmp') as temp_dir:
+        token = _TEMP_DIR.set(temp_dir)
+        try:
+            yield temp_dir
+        finally:
+            _TEMP_DIR.reset(token)
+
+
+# Keep the function signature same as tempfile.mkdtemp.
+# pylint: disable=redefined-builtin
+def mkdtemp(suffix: Optional[str] = None,
+            prefix: Optional[str] = None,
+            dir: Optional[str] = None) -> str:
+    """Create a temporary directory in the temp dir of current context.
+
+    The directory will be cleaned when the current context exits.
+    If there is no temp dir in current context, this function is equivalent to
+    tempfile.mkdtemp.
+    """
+    context_temp_dir = _TEMP_DIR.get()
+
+    if context_temp_dir is not None and dir is None:
+        dir = context_temp_dir
+    elif context_temp_dir is not None and dir is not None:
+        dir = os.path.join(context_temp_dir, dir)
+        os.makedirs(dir, exist_ok=True)
+
+    return tempfile.mkdtemp(suffix=suffix, prefix=prefix, dir=dir)


Test patch:
diff --git a/tests/common_test_fixtures.py b/tests/common_test_fixtures.py
index 302dd8a24..141c7360c 100644
--- a/tests/common_test_fixtures.py
+++ b/tests/common_test_fixtures.py
@@ -24,6 +24,9 @@ from sky.provision.aws import config as aws_config
 from sky.provision.kubernetes import utils as kubernetes_utils
 from sky.serve import serve_state
 from sky.server import common as server_common
+from sky.server import constants as server_constants
+from sky.server import rest
+from sky.server import versions
 from sky.server.requests import executor
 from sky.server.requests import requests as api_requests
 from sky.server.server import app
@@ -109,12 +112,16 @@ def mock_client_requests(monkeypatch: pytest.MonkeyPatch, mock_queue,
                 # Add iter_content method to response
                 response.iter_content = iter_content
 
+            response.headers[server_constants.API_VERSION_HEADER] = str(
+                server_constants.API_VERSION)
+            response.headers[server_constants.VERSION_HEADER] = \
+                versions.get_local_readable_version()
             return response
         else:
             return original_func(url, *args, **kwargs)
 
-    # Use monkeypatch to replace `requests.post` and `requests.get`
-    monkeypatch.setattr(requests, 'request', mock_http_request)
+    # pylint: disable=protected-access
+    monkeypatch.setattr(rest._session, "request", mock_http_request)
 
 
 # Define helper functions at module level for pickleability
diff --git a/tests/load_tests/test_load_on_server.py b/tests/load_tests/test_load_on_server.py
index 11e532319..8497e231a 100644
--- a/tests/load_tests/test_load_on_server.py
+++ b/tests/load_tests/test_load_on_server.py
@@ -233,8 +233,13 @@ def test_validate_api(num_requests):
     run_concurrent_api_requests(num_requests, validate, 'API /validate')
 
 
+def test_api_status(num_requests):
+    print(f"Testing {num_requests} API status requests")
+    run_concurrent_api_requests(num_requests, sdk.api_status, 'API /status')
+
+
 all_requests = ['launch', 'status', 'logs', 'jobs', 'serve']
-all_apis = ['status', 'cli_status', 'tail_logs', 'validate']
+all_apis = ['status', 'cli_status', 'tail_logs', 'validate', 'api_status']
 
 if __name__ == '__main__':
     parser = argparse.ArgumentParser()
@@ -319,6 +324,9 @@ if __name__ == '__main__':
             elif api == 'validate':
                 thread = threading.Thread(target=test_validate_api,
                                           args=(args.n,))
+            elif api == 'api_status':
+                thread = threading.Thread(target=test_api_status,
+                                          args=(args.n,))
             if thread:
                 test_threads.append(thread)
                 thread.start()
diff --git a/tests/smoke_tests/backward_compat/test_backward_compat.py b/tests/smoke_tests/backward_compat/test_backward_compat.py
index df18cd723..22c5ebdba 100644
--- a/tests/smoke_tests/backward_compat/test_backward_compat.py
+++ b/tests/smoke_tests/backward_compat/test_backward_compat.py
@@ -23,7 +23,9 @@ class TestBackwardCompatibility:
     SERVE_PREFIX = 'test-back-compat'
     TEST_TIMEOUT = 1800  # 30 minutes
     BASE_API_VERSION = constants.API_VERSION
+    BASE_MIN_COMPATIBLE_API_VERSION = constants.MIN_COMPATIBLE_API_VERSION
     CURRENT_API_VERSION = constants.API_VERSION
+    CURRENT_MIN_COMPATIBLE_API_VERSION = constants.MIN_COMPATIBLE_API_VERSION
 
     # Environment paths
     BASE_ENV_DIR = pathlib.Path(
@@ -128,13 +130,48 @@ class TestBackwardCompatibility:
             'uv pip install -e .[all]',)
 
         base_sky_api_version = subprocess.run(
-            f'{self.ACTIVATE_BASE} && python -c "from sky.server import constants; print(constants.API_VERSION)"',
+            f'{self.ACTIVATE_BASE} && python -c "'
+            'from sky.server import constants; '
+            'print(f\'API_VERSION: {constants.API_VERSION}\'); '
+            # Handle the base that does not have MIN_COMPATIBLE_API_VERSION defined
+            'min_compatible_version = getattr(constants, \'MIN_COMPATIBLE_API_VERSION\', constants.API_VERSION); '
+            'print(f\'MIN_COMPATIBLE_API_VERSION: {min_compatible_version}\');'
+            '"',
             shell=True,
             check=False,
+            executable='/bin/bash',
             text=True,
             capture_output=True)
-        TestBackwardCompatibility.BASE_API_VERSION = base_sky_api_version.stdout.strip(
-        )
+        if base_sky_api_version.returncode != 0:
+            raise RuntimeError(
+                f'Failed to get base API version information. '
+                f'Return code: {base_sky_api_version.returncode}, '
+                f'stderr: {base_sky_api_version.stderr}, '
+                f'stdout: {base_sky_api_version.stdout}')
+
+        # Use regex to extract version numbers from the output
+        output = base_sky_api_version.stdout.strip()
+
+        # Extract API_VERSION
+        api_version_match = re.search(r'API_VERSION:\s*(\d+)', output)
+        if not api_version_match:
+            raise RuntimeError(f'Could not find API_VERSION in output. '
+                               f'Output: {output}, '
+                               f'stderr: {base_sky_api_version.stderr}')
+
+        # Extract MIN_COMPATIBLE_API_VERSION
+        min_compatible_version_match = re.search(
+            r'MIN_COMPATIBLE_API_VERSION:\s*(\d+)', output)
+        if not min_compatible_version_match:
+            raise RuntimeError(
+                f'Could not find MIN_COMPATIBLE_API_VERSION in output. '
+                f'Output: {output}, '
+                f'stderr: {base_sky_api_version.stderr}')
+
+        TestBackwardCompatibility.BASE_API_VERSION = int(
+            api_version_match.group(1))
+        TestBackwardCompatibility.BASE_MIN_COMPATIBLE_API_VERSION = int(
+            min_compatible_version_match.group(1))
 
         yield  # Optional teardown logic
         self._run_cmd(f'{self.ACTIVATE_CURRENT} && sky api stop',)
@@ -384,11 +421,19 @@ class TestBackwardCompatibility:
 
     def test_client_server_compatibility(self, generic_cloud: str):
         """Test client server compatibility across versions"""
-        if self.BASE_API_VERSION != self.CURRENT_API_VERSION:
-            pytest.skip(
+        if self.BASE_API_VERSION < self.CURRENT_MIN_COMPATIBLE_API_VERSION or \
+                self.CURRENT_API_VERSION < self.BASE_MIN_COMPATIBLE_API_VERSION:
+            # This test runs against the master branch or the latest release
+            # version, which must enforce compatibility in this test based on
+            # our new version strategy that adjacent minor versions must be
+            # compatible.
+            pytest.fail(
                 f'Base API version: {self.BASE_API_VERSION} and current API '
-                f'version: {self.CURRENT_API_VERSION} are different, '
-                'skipping test')
+                f'version: {self.CURRENT_API_VERSION} are not compatible:\n'
+                f'- Base minimal compatible API version: {self.BASE_MIN_COMPATIBLE_API_VERSION}\n'
+                f'- Current minimal compatible API version: {self.CURRENT_MIN_COMPATIBLE_API_VERSION}\n'
+                'Change is rejected since it breaks the compatibility between adjacent versions'
+            )
         cluster_name = smoke_tests_utils.get_cluster_name()
         job_name = f"{cluster_name}-job"
         commands = [
diff --git a/tests/smoke_tests/smoke_tests_utils.py b/tests/smoke_tests/smoke_tests_utils.py
index a80525345..747ab3740 100644
--- a/tests/smoke_tests/smoke_tests_utils.py
+++ b/tests/smoke_tests/smoke_tests_utils.py
@@ -799,7 +799,7 @@ def get_response_from_request_id(request_id: str) -> Any:
     request_task = None
     if response.status_code == 200:
         request_task = requests_lib.Request.decode(
-            requests_lib.RequestPayload(**response.json()))
+            payloads.RequestPayload(**response.json()))
         return request_task.get_return_value()
     raise RuntimeError(f'Failed to get request {request_id}: '
                        f'{response.status_code} {response.text}')
diff --git a/tests/test_api_compatibility.py b/tests/test_api_compatibility.py
index 4d2b5890c..10a5c18cc 100644
--- a/tests/test_api_compatibility.py
+++ b/tests/test_api_compatibility.py
@@ -35,20 +35,28 @@ def fetch_master_api_version() -> str:
     with urllib.request.urlopen(MASTER_CONSTANTS_URL) as response:
         master_content = response.read().decode('utf-8')
 
+    return parse_api_version(master_content)
+
+
+def parse_api_version(content: str) -> str:
+    """Parse the API version from the content."""
     # Extract API_VERSION using regex
-    # Matches: API_VERSION = 'value' (pylint enforced format)
-    pattern = r"API_VERSION = '([^']+)'"
-    match = re.search(pattern, master_content)
+    # Matches: API_VERSION = value (pylint enforced format)
+    pattern = r"API_VERSION\s=\s(\d+)"
+    match = re.search(pattern, content)
 
     if match:
         return match.group(1)
     else:
-        raise ValueError("API_VERSION not found in master constants.py")
+        raise ValueError("API_VERSION not found in constants.py")
 
 
 def check_api_version_compatibility() -> bool:
     """Check if current and master API versions match."""
-    current_version = current_constants.API_VERSION
+    # Parse instead of read the constants directly to avoid regression breaking
+    # the test when the format on PR branch is changed.
+    current_content = inspect.getsource(current_constants)
+    current_version = parse_api_version(current_content)
     master_version = fetch_master_api_version()
 
     return current_version == master_version
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 83297d9ae..c9e361251 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -109,33 +109,6 @@ class TestWithNoCloudEnabled:
         assert isinstance(result.exception, ValueError)
 
 
-class TestServerVersion:
-
-    def test_cli_low_version_server_high_version(self, monkeypatch,
-                                                 mock_client_requests):
-        mock_server_api_version(monkeypatch, '2')
-        monkeypatch.setattr(server.constants, 'API_VERSION', 3)
-        cli_runner = cli_testing.CliRunner()
-
-        result = cli_runner.invoke(command.status, [])
-        assert "Client and local API server version mismatch" in str(
-            result.exception)
-        assert result.exit_code == 1
-
-    def test_cli_high_version_server_low_version(self, monkeypatch,
-                                                 mock_client_requests):
-        mock_server_api_version(monkeypatch, '3')
-        monkeypatch.setattr(server.constants, 'API_VERSION', 2)
-        cli_runner = cli_testing.CliRunner()
-
-        result = cli_runner.invoke(command.status, [])
-
-        # Verify the error message contains correct versions
-        assert "Client and local API server version mismatch" in str(
-            result.exception)
-        assert result.exit_code == 1
-
-
 class TestHelperFunctions:
 
     def test_get_cluster_records_and_set_ssh_config(self, monkeypatch):
diff --git a/tests/unit_tests/test_sky/server/test_common.py b/tests/unit_tests/test_sky/server/test_common.py
index e46e8810f..1aa7b1fa9 100644
--- a/tests/unit_tests/test_sky/server/test_common.py
+++ b/tests/unit_tests/test_sky/server/test_common.py
@@ -81,7 +81,7 @@ def test_local_client_server_mismatch(mock_is_local, mock_get_status):
             common.check_server_healthy()
 
         # Correct error message
-        assert 'Client and local API server version mismatch' in str(
+        assert 'The local SkyPilot API server is not compatible with the client' in str(
             exc_info.value)
         # Should hint user to restart local API server
         assert 'sky api stop; sky api start' in str(exc_info.value)
@@ -121,17 +121,14 @@ def test_remote_server_older(mock_is_local, mock_get_status):
         status=ApiServerStatus.VERSION_MISMATCH,
         api_version='0',
         version='1.0.0-dev20250415',
-        commit='abc123')
+        commit='abc123',
+        error='SkyPilot API server is too old')
 
     with pytest.raises(RuntimeError) as exc_info:
         common.check_server_healthy()
 
     # Correct error message
     assert 'SkyPilot API server is too old' in str(exc_info.value)
-    # Should hint user to upgrade remote server
-    assert 'Contact your administrator to upgrade the remote API server' in str(
-        exc_info.value)
-    assert 'or downgrade your local client with' in str(exc_info.value)
 
 
 @mock.patch('sky.server.common.get_api_server_status')
@@ -143,66 +140,14 @@ def test_client_older(mock_is_local, mock_get_status):
         status=ApiServerStatus.VERSION_MISMATCH,
         api_version=str(sys.maxsize),
         version='1.0.0-dev20250415',
-        commit='abc123')
+        commit='abc123',
+        error='Your SkyPilot client is too old')
 
     with pytest.raises(RuntimeError) as exc_info:
         common.check_server_healthy()
 
     # Correct error message
     assert 'Your SkyPilot client is too old' in str(exc_info.value)
-    # Should hint user to upgrade client
-    assert 'Upgrade your client with' in str(exc_info.value)
-
-
-def test_get_version_info_hint():
-    """Test the version info hint."""
-    # Test dev version
-    server_info = ApiServerInfo(status=ApiServerStatus.VERSION_MISMATCH,
-                                api_version='1',
-                                version='1.0.0-dev0',
-                                commit='abc123')
-    with mock.patch('sky.__version__', '1.0.0-dev0'), \
-         mock.patch('sky.__commit__', 'def456'):
-        hint = common._get_version_info_hint(server_info)
-        assert 'client version: v1.0.0-dev0 with commit def456' in hint
-        assert 'server version: v1.0.0-dev0 with commit abc123' in hint
-
-    # Test stable version
-    server_info = ApiServerInfo(status=ApiServerStatus.VERSION_MISMATCH,
-                                api_version='1',
-                                version='1.0.0',
-                                commit='abc123')
-    with mock.patch('sky.__version__', '1.1.0'):
-        hint = common._get_version_info_hint(server_info)
-        assert 'client version: v1.1.0' in hint
-        assert 'server version: v1.0.0' in hint
-
-
-def test_install_server_version_command():
-    """Test the install server version command."""
-    # Test dev version
-    server_info = ApiServerInfo(status=ApiServerStatus.VERSION_MISMATCH,
-                                api_version='1',
-                                version='1.0.0-dev0',
-                                commit='abc123')
-    cmd = common._install_server_version_command(server_info)
-    assert cmd == 'pip install git+https://github.com/skypilot-org/skypilot@abc123'
-
-    # Test nightly version
-    server_info = ApiServerInfo(status=ApiServerStatus.VERSION_MISMATCH,
-                                api_version='1',
-                                version='1.0.0-dev20250415',
-                                commit='abc123')
-    cmd = common._install_server_version_command(server_info)
-    assert cmd == 'pip install -U "skypilot-nightly==1.0.0-dev20250415"'
-
-    # Test stable version
-    server_info = ApiServerInfo(status=ApiServerStatus.VERSION_MISMATCH,
-                                api_version='1',
-                                version='1.0.0',
-                                commit='abc123')
-    cmd = common._install_server_version_command(server_info)
-    assert cmd == 'pip install -U "skypilot==1.0.0"'
 
 
 @pytest.fixture
diff --git a/tests/unit_tests/test_sky/server/test_rest.py b/tests/unit_tests/test_sky/server/test_rest.py
index 4582942cd..4aedf46fd 100644
--- a/tests/unit_tests/test_sky/server/test_rest.py
+++ b/tests/unit_tests/test_sky/server/test_rest.py
@@ -186,67 +186,52 @@ class TestRetryOnServerUnavailableDecorator:
 class TestRestRequestFunctions:
     """Test cases for REST request functions (request and request_without_retry)."""
 
-    @mock.patch('sky.adaptors.common.LazyImport')
-    def test_request_successful_post(self, mock_lazy_import):
+    @mock.patch('sky.server.rest._session')
+    def test_post_successful_request(self, mock_session):
         """Test successful POST request."""
-        mock_requests = mock.Mock()
         mock_response = mock.Mock()
         mock_response.status_code = 200
-        mock_requests.request.return_value = mock_response
-        mock_lazy_import.return_value = mock_requests
-
-        # Import after mocking to get the mocked requests
-        import importlib
-        importlib.reload(rest)
+        mock_response.headers = {}
+        mock_session.request.return_value = mock_response
 
         result = rest.request('POST', 'http://test.com', json={'key': 'value'})
 
         assert result == mock_response
-        mock_requests.request.assert_called_once_with('POST',
-                                                      'http://test.com',
-                                                      json={'key': 'value'})
+        mock_session.request.assert_called_once_with('POST',
+                                                     'http://test.com',
+                                                     json={'key': 'value'})
 
-    @mock.patch('sky.adaptors.common.LazyImport')
-    def test_request_successful_get(self, mock_lazy_import):
+    @mock.patch('sky.server.rest._session')
+    def test_get_successful_request(self, mock_session):
         """Test successful GET request."""
-        mock_requests = mock.Mock()
         mock_response = mock.Mock()
         mock_response.status_code = 200
-        mock_requests.request.return_value = mock_response
-        mock_lazy_import.return_value = mock_requests
-
-        # Import after mocking to get the mocked requests
-        import importlib
-        importlib.reload(rest)
+        mock_response.headers = {}
+        mock_session.request.return_value = mock_response
 
         result = rest.request('GET',
                               'http://test.com',
                               params={'param': 'value'})
 
         assert result == mock_response
-        mock_requests.request.assert_called_once_with('GET',
-                                                      'http://test.com',
-                                                      params={'param': 'value'})
-
-    @mock.patch('sky.adaptors.common.LazyImport')
-    def test_request_with_503_response_retries(self, mock_lazy_import):
-        """Test request retries on 503 response."""
-        mock_requests = mock.Mock()
+        mock_session.request.assert_called_once_with('GET',
+                                                     'http://test.com',
+                                                     params={'param': 'value'})
 
+    @mock.patch('sky.server.rest._session')
+    def test_post_with_503_response_retries(self, mock_session):
+        """Test POST request retries on 503 response."""
         # First call returns 503, second call succeeds
         mock_response_503 = mock.Mock()
         mock_response_503.status_code = 503
+        mock_response_503.headers = {}
         mock_response_200 = mock.Mock()
         mock_response_200.status_code = 200
+        mock_response_200.headers = {}
 
-        mock_requests.request.side_effect = [
+        mock_session.request.side_effect = [
             mock_response_503, mock_response_200
         ]
-        mock_lazy_import.return_value = mock_requests
-
-        # Import after mocking to get the mocked requests
-        import importlib
-        importlib.reload(rest)
 
         with mock.patch('time.sleep'):  # Speed up test
             result = rest.request('POST',
@@ -254,40 +239,33 @@ class TestRestRequestFunctions:
                                   json={'key': 'value'})
 
         assert result == mock_response_200
-        assert mock_requests.request.call_count == 2
+        assert mock_session.request.call_count == 2
 
-    @mock.patch('sky.adaptors.common.LazyImport')
-    def test_request_without_retry_no_retry_on_503(self, mock_lazy_import):
-        """Test request_without_retry does not retry on 503 response."""
-        mock_requests = mock.Mock()
+    @mock.patch('sky.server.rest._session')
+    def test_get_with_503_response_retries(self, mock_session):
+        """Test GET request retries on 503 response."""
+        # First call returns 503, second call succeeds
         mock_response_503 = mock.Mock()
         mock_response_503.status_code = 503
+        mock_response_503.headers = {}
+        mock_response_200 = mock.Mock()
+        mock_response_200.status_code = 200
+        mock_response_200.headers = {}
 
-        mock_requests.request.return_value = mock_response_503
-        mock_lazy_import.return_value = mock_requests
-
-        # Import after mocking to get the mocked requests
-        import importlib
-        importlib.reload(rest)
+        mock_session.request.side_effect = [
+            mock_response_503, mock_response_200
+        ]
 
         with pytest.raises(exceptions.ServerTemporarilyUnavailableError):
             rest.request_without_retry('GET', 'http://test.com')
 
-        # Should only be called once (no retry)
-        assert mock_requests.request.call_count == 1
-
-    @mock.patch('sky.adaptors.common.LazyImport')
-    def test_request_passes_through_kwargs(self, mock_lazy_import):
-        """Test that request passes through additional kwargs."""
-        mock_requests = mock.Mock()
+    @mock.patch('sky.server.rest._session')
+    def test_post_passes_through_kwargs(self, mock_session):
+        """Test that POST passes through additional kwargs."""
         mock_response = mock.Mock()
         mock_response.status_code = 200
-        mock_requests.request.return_value = mock_response
-        mock_lazy_import.return_value = mock_requests
-
-        # Import after mocking to get the mocked requests
-        import importlib
-        importlib.reload(rest)
+        mock_response.headers = {}
+        mock_session.request.return_value = mock_response
 
         result = rest.request('POST',
                               'http://test.com',
@@ -296,26 +274,20 @@ class TestRestRequestFunctions:
                               timeout=30)
 
         assert result == mock_response
-        mock_requests.request.assert_called_once_with(
+        mock_session.request.assert_called_once_with(
             'POST',
             'http://test.com',
             json={'key': 'value'},
             headers={'Authorization': 'Bearer token'},
             timeout=30)
 
-    @mock.patch('sky.adaptors.common.LazyImport')
-    def test_request_without_retry_passes_through_kwargs(
-            self, mock_lazy_import):
-        """Test that request_without_retry passes through additional kwargs."""
-        mock_requests = mock.Mock()
+    @mock.patch('sky.server.rest._session')
+    def test_get_passes_through_kwargs(self, mock_session):
+        """Test that GET passes through additional kwargs."""
         mock_response = mock.Mock()
         mock_response.status_code = 200
-        mock_requests.request.return_value = mock_response
-        mock_lazy_import.return_value = mock_requests
-
-        # Import after mocking to get the mocked requests
-        import importlib
-        importlib.reload(rest)
+        mock_response.headers = {}
+        mock_session.request.return_value = mock_response
 
         result = rest.request_without_retry('GET',
                                             'http://test.com',
@@ -324,7 +296,7 @@ class TestRestRequestFunctions:
                                             timeout=30)
 
         assert result == mock_response
-        mock_requests.request.assert_called_once_with(
+        mock_session.request.assert_called_once_with(
             'GET',
             'http://test.com',
             params={'param': 'value'},
diff --git a/tests/unit_tests/test_sky/server/test_server.py b/tests/unit_tests/test_sky/server/test_server.py
index d4a9ce104..e28bee88c 100644
--- a/tests/unit_tests/test_sky/server/test_server.py
+++ b/tests/unit_tests/test_sky/server/test_server.py
@@ -212,3 +212,49 @@ async def test_logs():
             request_id=mock.ANY,
             logs_path=mock_request_task.log_path,
             background_tasks=mock_background_tasks)
+
+
+@mock.patch('sky.utils.context_utils.hijack_sys_attrs')
+@mock.patch('asyncio.run')
+def test_server_run_uses_uvloop(mock_asyncio_run, mock_hijack_sys_attrs):
+    """Test that Server.run uses uvloop event loop policy."""
+    from sky.server.uvicorn import Server
+
+    config = uvicorn.Config(app='sky.server.server:app',
+                            host='127.0.0.1',
+                            port=8000)
+    server_instance = Server(config)
+    original_setup = config.setup_event_loop
+
+    uvloop_policy_set = False
+    uvloop_available = True
+
+    def setup_and_check():
+        # Call original setup to configure event loop
+        original_setup()
+        # Check if uvloop policy is now set
+        nonlocal uvloop_policy_set, uvloop_available
+        import asyncio
+        try:
+            import uvloop
+            policy = asyncio.get_event_loop_policy()
+            uvloop_policy_set = isinstance(policy, uvloop.EventLoopPolicy)
+        except ImportError:
+            # uvloop not available
+            uvloop_available = False
+
+    with mock.patch.object(config,
+                           'setup_event_loop',
+                           side_effect=setup_and_check):
+        # Call server.run
+        server_instance.run()
+
+    mock_asyncio_run.assert_called_once()
+
+    # Check uvloop policy was set (if uvloop is available)
+    if uvloop_available:
+        assert uvloop_policy_set, (
+            "Expected uvloop event loop policy to be set when uvloop "
+            "is available")
+    else:
+        pytest.skip("uvloop not available, skipping uvloop policy check")
diff --git a/tests/unit_tests/test_sky/server/test_versions.py b/tests/unit_tests/test_sky/server/test_versions.py
new file mode 100644
index 000000000..382e496eb
--- /dev/null
+++ b/tests/unit_tests/test_sky/server/test_versions.py
@@ -0,0 +1,409 @@
+"""Tests for sky.server.versions module."""
+
+from unittest import mock
+
+import pytest
+
+from sky import exceptions
+from sky.server import constants
+from sky.server import versions
+
+
+def test_check_version_compatibility_compatible_versions():
+    """Test check_version_compatibility with compatible versions."""
+    headers = {
+        constants.API_VERSION_HEADER: '1',
+        constants.VERSION_HEADER: '1.0.0'
+    }
+
+    with mock.patch.object(constants, 'MIN_COMPATIBLE_API_VERSION', 1):
+        result = versions.check_compatibility_at_server(headers)
+
+    assert result is not None
+    assert result.api_version == 1
+    assert result.version == '1.0.0'
+    assert result.error is None
+
+
+def test_check_version_compatibility_missing_headers():
+    """Test check_version_compatibility with missing headers."""
+    # Test missing API version header - should return None for backward compatibility
+    headers = {constants.VERSION_HEADER: '1.0.0'}
+    result = versions.check_compatibility_at_server(headers)
+    assert result is None
+
+    # Test missing version header - should return None for backward compatibility
+    headers = {constants.API_VERSION_HEADER: '1'}
+    result = versions.check_compatibility_at_server(headers)
+    assert result is None
+
+
+def test_check_version_compatibility_headers_with_none_values():
+    """Test check_version_compatibility with headers that contain None values."""
+    # Test with headers where values are None
+    headers = {
+        constants.API_VERSION_HEADER: None,
+        constants.VERSION_HEADER: '1.0.0'
+    }
+    result = versions.check_compatibility_at_server(headers)
+    assert result is None
+
+    headers = {
+        constants.API_VERSION_HEADER: '1',
+        constants.VERSION_HEADER: None
+    }
+    result = versions.check_compatibility_at_server(headers)
+    assert result is None
+
+
+def test_check_version_compatibility_invalid_api_version():
+    """Test check_version_compatibility with invalid API version."""
+    headers = {
+        constants.API_VERSION_HEADER: 'invalid',
+        constants.VERSION_HEADER: '1.0.0'
+    }
+
+    with pytest.raises(ValueError, match='is not a valid API version'):
+        versions.check_compatibility_at_server(headers)
+
+
+def test_check_version_compatibility_incompatible_client():
+    """Test check_version_compatibility with incompatible client."""
+    headers = {
+        constants.API_VERSION_HEADER: '1',
+        constants.VERSION_HEADER: '0.9.0'
+    }
+
+    with mock.patch.object(constants, 'MIN_COMPATIBLE_API_VERSION', 2), \
+         mock.patch.object(constants, 'MIN_COMPATIBLE_VERSION', '1.0.0'), \
+         mock.patch('sky.server.versions.get_local_readable_version',
+                    return_value='1.0.0'), \
+         mock.patch('sky.server.versions.install_version_command',
+                    return_value='pip install skypilot==0.9.0'):
+
+        result = versions.check_compatibility_at_server(headers)
+
+    assert result is not None
+    assert result.api_version == 1
+    assert result.version == '0.9.0'
+    assert result.error is not None
+    assert 'client version is too old' in result.error
+
+
+def test_check_version_compatibility_incompatible_server():
+    """Test check_version_compatibility with incompatible server."""
+    headers = {
+        constants.API_VERSION_HEADER: '1',
+        constants.VERSION_HEADER: '0.9.0'
+    }
+
+    with mock.patch.object(constants, 'MIN_COMPATIBLE_API_VERSION', 2), \
+         mock.patch.object(constants, 'MIN_COMPATIBLE_VERSION', '1.0.0'), \
+         mock.patch('sky.server.versions.get_local_readable_version',
+                    return_value='1.0.0'), \
+         mock.patch('sky.server.versions.parse_readable_version',
+                    return_value=('0.9.0', None)), \
+         mock.patch('sky.server.versions.install_version_command',
+                    return_value='pip install skypilot==0.9.0'):
+
+        result = versions.check_compatibility_at_client(headers)
+
+    assert result is not None
+    assert result.api_version == 1
+    assert result.version == '0.9.0'
+    assert result.error is not None
+    assert 'server version is too old' in result.error
+
+
+def test_get_local_readable_version_dev():
+    """Test get_local_readable_version with dev version."""
+    with mock.patch('sky.__version__', versions.DEV_VERSION), \
+         mock.patch('sky.__commit__', 'abc123'):
+
+        result = versions.get_local_readable_version()
+
+    assert result == f'{versions.DEV_VERSION} (commit: abc123)'
+
+
+def test_get_local_readable_version_regular():
+    """Test get_local_readable_version with regular version."""
+    with mock.patch('sky.__version__', '1.2.3'):
+        result = versions.get_local_readable_version()
+
+    assert result == '1.2.3'
+
+
+def test_parse_readable_version_with_commit():
+    """Test parse_readable_version with commit info."""
+    version = '1.0.0-dev0 (commit: abc123)'
+    base_version, commit = versions.parse_readable_version(version)
+
+    assert base_version == '1.0.0-dev0'
+    assert commit == 'abc123'
+
+
+def test_parse_readable_version_without_commit():
+    """Test parse_readable_version without commit info."""
+    version = '1.2.3'
+    base_version, commit = versions.parse_readable_version(version)
+
+    assert base_version == '1.2.3'
+    assert commit is None
+
+
+def test_parse_readable_version_edge_cases():
+    """Test parse_readable_version with edge cases."""
+    # Version with special characters but no commit
+    version = '1.2.3-alpha1'
+    base_version, commit = versions.parse_readable_version(version)
+    assert base_version == '1.2.3-alpha1'
+    assert commit is None
+
+    # Empty version
+    version = ''
+    base_version, commit = versions.parse_readable_version(version)
+    assert base_version == ''
+    assert commit is None
+
+
+def test_install_version_command_dev_with_commit():
+    """Test install_version_command with dev version and commit."""
+    result = versions.install_version_command(versions.DEV_VERSION, 'abc123')
+    expected = 'pip install git+https://github.com/skypilot-org/skypilot@abc123'
+    assert result == expected
+
+
+def test_install_version_command_dev_without_commit():
+    """Test install_version_command with dev version but no commit."""
+    result = versions.install_version_command(versions.DEV_VERSION)
+    # Should fall through to regular version case since no commit provided
+    expected = f'pip install -U "skypilot=={versions.DEV_VERSION}"'
+    assert result == expected
+
+
+def test_install_version_command_nightly():
+    """Test install_version_command with nightly version."""
+    result = versions.install_version_command('1.2.3-dev1')
+    expected = 'pip install -U "skypilot-nightly==1.2.3-dev1"'
+    assert result == expected
+
+
+def test_install_version_command_regular():
+    """Test install_version_command with regular version."""
+    result = versions.install_version_command('1.2.3')
+    expected = 'pip install -U "skypilot==1.2.3"'
+    assert result == expected
+
+
+def test_remind_minor_version_upgrade_should_remind():
+    """Test _remind_minor_version_upgrade when upgrade is needed."""
+    # Reset the global flag
+    versions._reminded_for_minor_version_upgrade = False
+
+    with mock.patch('sky.__version__', '1.0.0'), \
+         mock.patch('sky.server.versions.logger') as mock_logger, \
+         mock.patch('sky.server.versions.install_version_command',
+                    return_value='pip install skypilot==1.1.0'):
+
+        versions._remind_minor_version_upgrade('1.1.0')
+
+    mock_logger.warning.assert_called_once()
+    assert versions._reminded_for_minor_version_upgrade is True
+
+
+def test_remind_minor_version_upgrade_skip_dev_versions():
+    """Test _remind_minor_version_upgrade skips dev versions."""
+    versions._reminded_for_minor_version_upgrade = False
+
+    with mock.patch('sky.__version__', '1.0.0-dev0'), \
+         mock.patch('sky.server.versions.logger') as mock_logger:
+
+        versions._remind_minor_version_upgrade('1.1.0')
+
+    mock_logger.warning.assert_not_called()
+    assert versions._reminded_for_minor_version_upgrade is False
+
+
+def test_remind_minor_version_upgrade_skip_when_already_reminded():
+    """Test _remind_minor_version_upgrade skips when already reminded."""
+    versions._reminded_for_minor_version_upgrade = True
+
+    with mock.patch('sky.__version__', '1.0.0'), \
+         mock.patch('sky.server.versions.logger') as mock_logger:
+
+        versions._remind_minor_version_upgrade('1.1.0')
+
+    mock_logger.warning.assert_not_called()
+
+
+def test_remind_minor_version_upgrade_skip_same_version():
+    """Test _remind_minor_version_upgrade skips when versions are same."""
+    versions._reminded_for_minor_version_upgrade = False
+
+    with mock.patch('sky.__version__', '1.1.0'), \
+         mock.patch('sky.server.versions.logger') as mock_logger:
+
+        versions._remind_minor_version_upgrade('1.1.0')
+
+    mock_logger.warning.assert_not_called()
+
+
+def test_remind_minor_version_upgrade_skip_newer_local():
+    """Test _remind_minor_version_upgrade skips when local is newer."""
+    versions._reminded_for_minor_version_upgrade = False
+
+    with mock.patch('sky.__version__', '1.2.0'), \
+         mock.patch('sky.server.versions.logger') as mock_logger:
+
+        versions._remind_minor_version_upgrade('1.1.0')
+
+    mock_logger.warning.assert_not_called()
+
+
+def test_version_info_named_tuple():
+    """Test VersionInfo NamedTuple creation and access."""
+    # Test with all fields
+    version_info = versions.VersionInfo(api_version=1,
+                                        version='1.2.3',
+                                        error='test error')
+    assert version_info.api_version == 1
+    assert version_info.version == '1.2.3'
+    assert version_info.error == 'test error'
+
+    # Test with default error field
+    version_info = versions.VersionInfo(api_version=2, version='1.3.0')
+    assert version_info.api_version == 2
+    assert version_info.version == '1.3.0'
+    assert version_info.error is None
+
+
+def test_minimal_api_version_decorator_no_remote_version():
+    """Test minimal_api_version decorator when remote version is None."""
+
+    @versions.minimal_api_version(2)
+    def test_function(arg1, arg2=None):
+        return f"called with {arg1}, {arg2}"
+
+    # Mock get_remote_api_version to return None
+    with mock.patch('sky.server.versions.get_remote_api_version',
+                    return_value=None):
+        result = test_function("test", arg2="value")
+        assert result == "called with test, value"
+
+
+def test_minimal_api_version_decorator_compatible_version():
+    """Test minimal_api_version decorator when remote version is compatible."""
+
+    @versions.minimal_api_version(2)
+    def test_function(arg1):
+        return f"success: {arg1}"
+
+    # Mock get_remote_api_version to return a compatible version
+    with mock.patch('sky.server.versions.get_remote_api_version',
+                    return_value=3):
+        result = test_function("test")
+        assert result == "success: test"
+
+    # Test with exact minimum version
+    with mock.patch('sky.server.versions.get_remote_api_version',
+                    return_value=2):
+        result = test_function("test")
+        assert result == "success: test"
+
+
+def test_minimal_api_version_decorator_incompatible_dev_version():
+    """Test minimal_api_version decorator with incompatible version and dev client."""
+
+    @versions.minimal_api_version(3)
+    def test_function():
+        return "should not be called"
+
+    with mock.patch('sky.server.versions.get_remote_api_version',
+                    return_value=2), \
+         mock.patch('sky.server.versions.get_remote_version',
+                    return_value='1.0.0-dev0'), \
+         mock.patch('sky.__version__', '1.1.0-dev0'), \
+         mock.patch('sky.utils.ux_utils.print_exception_no_traceback'):
+
+        with pytest.raises(exceptions.APINotSupportedError) as exc_info:
+            test_function()
+
+        error_message = str(exc_info.value)
+        assert "test_function" in error_message
+        assert "Please upgrade the remote server." in error_message
+        assert "1.0.0-dev0" in error_message
+
+
+def test_minimal_api_version_decorator_incompatible_release_version():
+    """Test minimal_api_version decorator with incompatible version and release client."""
+
+    @versions.minimal_api_version(3)
+    def test_function(param1, param2):
+        return f"{param1} {param2}"
+
+    with mock.patch('sky.server.versions.get_remote_api_version',
+                    return_value=1), \
+         mock.patch('sky.server.versions.get_remote_version',
+                    return_value='0.9.0'), \
+         mock.patch('sky.__version__', '1.0.0'), \
+         mock.patch('sky.utils.ux_utils.print_exception_no_traceback'):
+
+        with pytest.raises(exceptions.APINotSupportedError) as exc_info:
+            test_function("arg1", "arg2")
+
+        error_message = str(exc_info.value)
+        assert "test_function" in error_message
+        assert "Upgrade the remote server to 1.0.0" in error_message
+        assert "0.9.0" in error_message
+
+
+def test_minimal_api_version_decorator_preserves_function_metadata():
+    """Test that minimal_api_version decorator preserves function metadata."""
+
+    @versions.minimal_api_version(1)
+    def test_function_with_docstring(arg1: str, arg2: int = 42) -> str:
+        """This is a test function with docstring."""
+        return f"{arg1}: {arg2}"
+
+    # Check that function metadata is preserved
+    assert test_function_with_docstring.__name__ == "test_function_with_docstring"
+    assert test_function_with_docstring.__doc__ == "This is a test function with docstring."
+
+    # Check that the function still works normally when version is compatible
+    with mock.patch('sky.server.versions.get_remote_api_version',
+                    return_value=1):
+        result = test_function_with_docstring("test")
+        assert result == "test: 42"
+
+
+def test_minimal_api_version_decorator_edge_case_zero_min_version():
+    """Test minimal_api_version decorator with minimum version 0."""
+
+    @versions.minimal_api_version(0)
+    def test_function():
+        return "called"
+
+    # Any version should be compatible with min version 0
+    with mock.patch('sky.server.versions.get_remote_api_version',
+                    return_value=0):
+        result = test_function()
+        assert result == "called"
+
+    with mock.patch('sky.server.versions.get_remote_api_version',
+                    return_value=1):
+        result = test_function()
+        assert result == "called"
+
+
+def test_minimal_api_version_decorator_function_with_kwargs():
+    """Test minimal_api_version decorator with function that has *args and **kwargs."""
+
+    @versions.minimal_api_version(1)
+    def test_function(*args, **kwargs):
+        return {"args": args, "kwargs": kwargs}
+
+    with mock.patch('sky.server.versions.get_remote_api_version',
+                    return_value=1):
+        result = test_function("arg1", "arg2", key1="value1", key2="value2")
+        assert result["args"] == ("arg1", "arg2")
+        assert result["kwargs"] == {"key1": "value1", "key2": "value2"}
diff --git a/tests/unit_tests/test_sky/utils/text_context.py b/tests/unit_tests/test_sky/utils/test_context.py
similarity index 100%
rename from tests/unit_tests/test_sky/utils/text_context.py
rename to tests/unit_tests/test_sky/utils/test_context.py
diff --git a/tests/unit_tests/test_sky/utils/test_tempstore.py b/tests/unit_tests/test_sky/utils/test_tempstore.py
new file mode 100644
index 000000000..9a0b450d2
--- /dev/null
+++ b/tests/unit_tests/test_sky/utils/test_tempstore.py
@@ -0,0 +1,282 @@
+"""Unit tests for sky.utils.tempstore module."""
+
+import os
+import pathlib
+import tempfile
+from unittest import mock
+
+import pytest
+
+from sky.utils import tempstore
+
+
+def test_tempdir_context_manager():
+    """Test tempdir context manager creates and cleans up temp directory."""
+    temp_path = None
+
+    with tempstore.tempdir() as temp_dir:
+        temp_path = pathlib.Path(temp_dir)
+        assert temp_path.exists()
+        assert temp_path.is_dir()
+        assert temp_path.name.startswith('sky-tmp')
+
+    # Directory should be cleaned up after context exits
+    assert not temp_path.exists()
+
+
+def test_tempdir_context_isolation():
+    """Test tempdir context manager provides isolated temp directories."""
+    path1 = None
+    path2 = None
+
+    with tempstore.tempdir() as temp_dir1:
+        path1 = temp_dir1
+        with tempstore.tempdir() as temp_dir2:
+            path2 = temp_dir2
+            # Should be different directories
+            assert path1 != path2
+            assert pathlib.Path(path1).exists()
+            assert pathlib.Path(path2).exists()
+
+        # Inner temp dir should be cleaned up
+        assert not pathlib.Path(path2).exists()
+        # Outer temp dir should still exist
+        assert pathlib.Path(path1).exists()
+
+    # Both should be cleaned up now
+    assert not pathlib.Path(path1).exists()
+    assert not pathlib.Path(path2).exists()
+
+
+def test_tempdir_nested_contexts():
+    """Test nested tempdir contexts work correctly."""
+    outer_dir = None
+    inner_dir = None
+
+    with tempstore.tempdir() as temp_dir1:
+        outer_dir = temp_dir1
+        # Create a temp dir inside the context
+        inner_temp = tempstore.mkdtemp()
+        assert inner_temp.startswith(outer_dir)
+
+        with tempstore.tempdir() as temp_dir2:
+            inner_dir = temp_dir2
+            # This should create temp dir in the new context
+            nested_temp = tempstore.mkdtemp()
+            assert nested_temp.startswith(inner_dir)
+            assert not nested_temp.startswith(outer_dir)
+
+        # Back to outer context
+        outer_temp = tempstore.mkdtemp()
+        assert outer_temp.startswith(outer_dir)
+
+
+def test_mkdtemp_no_context():
+    """Test mkdtemp without tempdir context."""
+    # Should work like regular tempfile.mkdtemp
+    temp_dir = tempstore.mkdtemp()
+    try:
+        assert os.path.exists(temp_dir)
+        assert os.path.isdir(temp_dir)
+        # Should not have sky-tmp prefix since no context
+        assert not os.path.basename(temp_dir).startswith('sky-tmp')
+    finally:
+        # Manual cleanup since no context
+        if os.path.exists(temp_dir):
+            os.rmdir(temp_dir)
+
+
+def test_mkdtemp_with_context():
+    """Test mkdtemp with tempdir context."""
+    with tempstore.tempdir() as context_dir:
+        temp_dir = tempstore.mkdtemp()
+
+        # Should be created inside context directory
+        assert temp_dir.startswith(context_dir)
+        assert os.path.exists(temp_dir)
+        assert os.path.isdir(temp_dir)
+
+        # Should be automatically cleaned up when context exits
+
+
+def test_mkdtemp_with_parameters():
+    """Test mkdtemp with suffix, prefix, and dir parameters."""
+    with tempstore.tempdir() as context_dir:
+        # Test with suffix and prefix
+        temp_dir = tempstore.mkdtemp(suffix='_test', prefix='test_')
+        assert os.path.basename(temp_dir).startswith('test_')
+        assert os.path.basename(temp_dir).endswith('_test')
+        assert temp_dir.startswith(context_dir)
+
+        # Test with explicit dir parameter
+        subdir = 'subdir'
+        temp_dir2 = tempstore.mkdtemp(dir=subdir)
+        expected_parent = os.path.join(context_dir, subdir)
+        assert temp_dir2.startswith(expected_parent)
+
+
+def test_mkdtemp_explicit_dir_no_context():
+    """Test mkdtemp with explicit dir but no context."""
+    with tempfile.TemporaryDirectory() as temp_base:
+        subdir = 'testsubdir'
+        full_subdir = os.path.join(temp_base, subdir)
+        os.makedirs(full_subdir)
+
+        temp_dir = tempstore.mkdtemp(dir=full_subdir)
+        try:
+            assert temp_dir.startswith(full_subdir)
+            assert os.path.exists(temp_dir)
+        finally:
+            if os.path.exists(temp_dir):
+                os.rmdir(temp_dir)
+
+
+def test_mkdtemp_explicit_dir_with_context():
+    """Test mkdtemp with explicit dir and context."""
+    with tempstore.tempdir() as context_dir:
+        subdir = 'mysubdir'
+        temp_dir = tempstore.mkdtemp(dir=subdir)
+
+        # Should be created in context_dir/subdir
+        expected_parent = os.path.join(context_dir, subdir)
+        assert temp_dir.startswith(expected_parent)
+        assert os.path.exists(temp_dir)
+
+        # Test nested directory creation
+        nested_dir = 'level1/level2/level3'
+        nested_temp = tempstore.mkdtemp(dir=nested_dir)
+        expected_nested = os.path.join(context_dir, nested_dir)
+        assert nested_temp.startswith(expected_nested)
+        assert os.path.exists(nested_temp)
+
+
+def test_mkdtemp_multiple_calls():
+    """Test multiple mkdtemp calls in same context create different dirs."""
+    with tempstore.tempdir() as context_dir:
+        temp_dir1 = tempstore.mkdtemp()
+        temp_dir2 = tempstore.mkdtemp()
+        temp_dir3 = tempstore.mkdtemp(prefix='special_')
+
+        # All should be different
+        assert temp_dir1 != temp_dir2 != temp_dir3
+
+        # All should exist
+        assert os.path.exists(temp_dir1)
+        assert os.path.exists(temp_dir2)
+        assert os.path.exists(temp_dir3)
+
+        # All should be in context directory
+        assert temp_dir1.startswith(context_dir)
+        assert temp_dir2.startswith(context_dir)
+        assert temp_dir3.startswith(context_dir)
+
+        # Special prefix should be honored
+        assert os.path.basename(temp_dir3).startswith('special_')
+
+
+def test_tempdir_exception_cleanup():
+    """Test tempdir cleans up even when exception occurs."""
+    temp_path = None
+
+    with pytest.raises(ValueError):
+        with tempstore.tempdir() as temp_dir:
+            temp_path = pathlib.Path(temp_dir)
+            assert temp_path.exists()
+            raise ValueError("Test exception")
+
+    # Directory should still be cleaned up despite exception
+    assert not temp_path.exists()
+
+
+def test_contextvar_isolation():
+    """Test that context variable is properly isolated."""
+    import threading
+
+    results = {}
+
+    def thread_func(thread_id):
+        with tempstore.tempdir() as temp_dir:
+            results[thread_id] = temp_dir
+            # Create a temp directory in this thread's context
+            thread_temp = tempstore.mkdtemp()
+            results[f'{thread_id}_temp'] = thread_temp
+
+    # Run in multiple threads
+    threads = []
+    for i in range(3):
+        thread = threading.Thread(target=thread_func, args=(i,))
+        threads.append(thread)
+        thread.start()
+
+    for thread in threads:
+        thread.join()
+
+    # Each thread should have different temp directories
+    assert len(set(results[i] for i in range(3))) == 3
+
+    # Each thread's mkdtemp should be in its own context
+    for i in range(3):
+        assert results[f'{i}_temp'].startswith(results[i])
+
+
+def test_tempdir_prefix():
+    """Test that tempdir uses correct prefix."""
+    with tempstore.tempdir() as temp_dir:
+        dir_name = os.path.basename(temp_dir)
+        assert dir_name.startswith('sky-tmp')
+
+
+def test_mkdtemp_with_none_parameters():
+    """Test mkdtemp handles None parameters correctly."""
+    with tempstore.tempdir() as context_dir:
+        # All None parameters should work
+        temp_dir = tempstore.mkdtemp(suffix=None, prefix=None, dir=None)
+        assert temp_dir.startswith(context_dir)
+        assert os.path.exists(temp_dir)
+
+
+def test_tempdir_returns_string():
+    """Test that tempdir yields a string path."""
+    with tempstore.tempdir() as temp_dir:
+        assert isinstance(temp_dir, str)
+        assert os.path.isabs(temp_dir)
+
+
+def test_mkdtemp_returns_string():
+    """Test that mkdtemp returns a string path."""
+    with tempstore.tempdir():
+        temp_dir = tempstore.mkdtemp()
+        assert isinstance(temp_dir, str)
+        assert os.path.isabs(temp_dir)
+
+
+def test_mkdtemp_creates_intermediate_dirs():
+    """Test that mkdtemp creates intermediate directories when needed."""
+    with tempstore.tempdir() as context_dir:
+        # Test deep nested directory creation
+        deep_dir = 'a/b/c/d/e'
+        temp_dir = tempstore.mkdtemp(dir=deep_dir)
+
+        # All intermediate directories should be created
+        intermediate_path = os.path.join(context_dir, deep_dir)
+        assert os.path.exists(intermediate_path)
+        assert temp_dir.startswith(intermediate_path)
+
+        # Verify we can create another temp dir in the same deep structure
+        temp_dir2 = tempstore.mkdtemp(dir=deep_dir)
+        assert temp_dir2.startswith(intermediate_path)
+        assert temp_dir != temp_dir2
+
+
+def test_mkdtemp_dir_with_existing_path():
+    """Test mkdtemp with dir parameter when path already exists."""
+    with tempstore.tempdir() as context_dir:
+        # Create a subdirectory first
+        subdir = 'existing_subdir'
+        full_subdir = os.path.join(context_dir, subdir)
+        os.makedirs(full_subdir)
+
+        # Should work fine even if directory already exists
+        temp_dir = tempstore.mkdtemp(dir=subdir)
+        assert temp_dir.startswith(full_subdir)
+        assert os.path.exists(temp_dir)