Repository: skypilot-org/skypilot
Pull request #4880 (2025-03-05 07:52:06)

Exclude `.pyc` and `__pycache__` files from config hash calculation to fix `test_launch_fast --kubernetes` failures
<!-- Describe the changes in this PR -->

# Issue

The `test_launch_fast --kubernetes` test has been failing intermittently on buildkite and occasionally on local machines. Upon investigation, it was discovered that the `config_hash` was changing due to modifications in `.pyc` files, specifically `sky/clouds/service_catalog/__pycache__/kubernetes_catalog.cpython-310.pyc`. This caused the `--fast` flag to not work as expected, as the cached wheel was not being reused.

[Failure build with detail log](https://buildkite.com/skypilot-1/smoke-tests/builds/290#01956494-3ed6-4dc9-8c4e-37883f35df0b)

# Change

* `_get_latest_modification_time` excludes `.pyc` and `__pycache__` files. This ensures that `build_sky_wheel` reuses the cached wheel if no changes are detected in the source code files.

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

# Test

- [x] `/smoke-test --kubernetes -k test_launch_fast` - [pass](https://buildkite.com/skypilot-1/smoke-tests/builds/291)
- [x] `/smoke-test --aws -k test_launch_fast` - [pass](https://buildkite.com/skypilot-1/smoke-tests/builds/292)
- [x] `/smoke-test --gcp -k test_launch_fast` - [pass](https://buildkite.com/skypilot-1/smoke-tests/builds/293)
- [x] `/smoke-test --azure -k test_launch_fast` - [pass](https://buildkite.com/skypilot-1/smoke-tests/builds/294)

zpoint: /smoke-test --kubernetes -k test_launch_fast
zpoint: /smoke-test --aws -k test_launch_fast
zpoint: /smoke-test --gcp -k test_launch_fast
zpoint: /smoke-test --azure -k test_launch_fast

Patch:
diff --git a/docs/source/getting-started/installation.rst b/docs/source/getting-started/installation.rst
index f1b9adbf8..fad747b1a 100644
--- a/docs/source/getting-started/installation.rst
+++ b/docs/source/getting-started/installation.rst
@@ -23,6 +23,7 @@ Install SkyPilot using pip:
 
           # Choose your cloud:
 
+          pip install "skypilot[kubernetes]"
           pip install "skypilot[aws]"
           pip install "skypilot[gcp]"
           pip install "skypilot[azure]"
@@ -35,7 +36,6 @@ Install SkyPilot using pip:
           pip install "skypilot[ibm]"
           pip install "skypilot[scp]"
           pip install "skypilot[vsphere]"
-          pip install "skypilot[kubernetes]"
           pip install "skypilot[nebius]"
           pip install "skypilot[all]"
 
@@ -52,6 +52,7 @@ Install SkyPilot using pip:
 
           # Choose your cloud:
 
+          pip install "skypilot-nightly[kubernetes]"
           pip install "skypilot-nightly[aws]"
           pip install "skypilot-nightly[gcp]"
           pip install "skypilot-nightly[azure]"
@@ -65,7 +66,6 @@ Install SkyPilot using pip:
           pip install "skypilot-nightly[ibm]"
           pip install "skypilot-nightly[scp]"
           pip install "skypilot-nightly[vsphere]"
-          pip install "skypilot-nightly[kubernetes]"
           pip install "skypilot-nightly[nebius]"
           pip install "skypilot-nightly[all]"
 
@@ -85,6 +85,7 @@ Install SkyPilot using pip:
 
           # Choose your cloud:
 
+          pip install -e ".[kubernetes]"
           pip install -e ".[aws]"
           pip install -e ".[gcp]"
           pip install -e ".[azure]"
@@ -97,7 +98,6 @@ Install SkyPilot using pip:
           pip install -e ".[ibm]"
           pip install -e ".[scp]"
           pip install -e ".[vsphere]"
-          pip install -e ".[kubernetes]"
           pip install -e ".[nebius]"
           pip install -e ".[all]"
 
@@ -231,7 +231,7 @@ See :ref:`SkyPilot on Kubernetes <kubernetes-overview>` for more.
 
 .. _aws-installation:
 
-Amazon Web Services (AWS)
+AWS
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 
@@ -251,7 +251,7 @@ To use AWS IAM Identity Center (AWS SSO), see :ref:`here<aws-sso>` for instructi
 
 .. _installation-gcp:
 
-Google Cloud Platform (GCP)
+GCP
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 .. code-block:: shell
@@ -292,7 +292,7 @@ Azure
 Hint: run ``az account subscription list`` to get a list of subscription IDs under your account.
 
 
-Oracle Cloud Infrastructure (OCI)
+OCI
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 To access Oracle Cloud Infrastructure (OCI), setup the credentials by following `this guide <https://docs.oracle.com/en-us/iaas/Content/API/Concepts/apisigningkey.htm>`__. After completing the steps in the guide, the :code:`~/.oci` folder should contain the following files:
@@ -440,10 +440,10 @@ Finally, install `rclone <https://rclone.org/>`_ via: ``curl https://rclone.org/
 
 
 
-Samsung Cloud Platform (SCP)
+SCP (Samsung Cloud Platform)
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-Samsung Cloud Platform(SCP) provides cloud services optimized for enterprise customers. You can learn more about SCP `here <https://cloud.samsungsds.com/>`__.
+Samsung Cloud Platform, or SCP, provides cloud services optimized for enterprise customers. You can learn more about SCP `here <https://cloud.samsungsds.com/>`__.
 
 To configure SCP access, you need access keys and the ID of the project your tasks will run. Go to the `Access Key Management <https://cloud.samsungsds.com/console/#/common/access-key-manage/list?popup=true>`_ page on your SCP console to generate the access keys, and the Project Overview page for the project ID. Then, add them to :code:`~/.scp/scp_credential` by running:
 
diff --git a/docs/source/reference/config.rst b/docs/source/reference/config.rst
index d926ef926..c786ce356 100644
--- a/docs/source/reference/config.rst
+++ b/docs/source/reference/config.rst
@@ -183,6 +183,8 @@ Supported bucket types:
 
 Configure resources for the managed jobs controller.
 
+For more details about tuning the jobs controller resources, see :ref:`jobs-controller-sizing`.
+
 Example:
 
 .. code-block:: yaml
@@ -190,10 +192,13 @@ Example:
   jobs:
     controller:
       resources:  # same spec as 'resources' in a task YAML
+        # optionally set specific cloud/region
         cloud: gcp
         region: us-central1
-        cpus: 4+  # number of vCPUs, max concurrent spot jobs = 2 * cpus
-        disk_size: 100
+        # default resources:
+        cpus: 4+
+        memory: 8x
+        disk_size: 50
 
 .. _config-yaml-allowed-clouds:
 
@@ -901,13 +906,13 @@ Default: ``10``.
 
 Autoscaler type (optional).
 
-Type of autoscaler to use.
+Type of autoscaler used by the underlying Kubernetes cluster. Used to configure the GPU labels used by the pods submitted by SkyPilot.
 
 Can be one of:
 
-- ``gke``: Google Kubernetes Engine Autopilot
-- ``eks``: Amazon EKS
-- ``aks``: Azure Kubernetes Service
+- ``gke``: Google Kubernetes Engine
+- ``karpenter``: Karpenter
+- ``generic``: Generic autoscaler, assumes nodes are labelled with ``skypilot.co/accelerator``.
 
 .. _config-yaml-kubernetes-pod-config:
 
diff --git a/docs/source/reference/kubernetes/kubernetes-getting-started.rst b/docs/source/reference/kubernetes/kubernetes-getting-started.rst
index 0b453041c..7d67ee9a5 100644
--- a/docs/source/reference/kubernetes/kubernetes-getting-started.rst
+++ b/docs/source/reference/kubernetes/kubernetes-getting-started.rst
@@ -348,9 +348,9 @@ FAQs
 
 * **Can I use multiple Kubernetes clusters with SkyPilot?**
 
-  SkyPilot can work with multiple Kubernetes contexts set in your kubeconfig file. By default, SkyPilot will use the current active context. To use a different context, change your current context using :code:`kubectl config use-context <context-name>`.
+  SkyPilot can work with multiple Kubernetes contexts in your kubeconfig file by setting the ``allowed_contexts`` key in :code:`~/.sky/config.yaml`. See :ref:`multi-kubernetes`.
 
-  If you would like to use multiple contexts seamlessly during failover, check out the :code:`allowed_contexts` feature in :ref:`config-yaml`.
+  If ``allowed_contexts`` is not set, SkyPilot will use the current active context. To use a different context, change your current context using :code:`kubectl config use-context <context-name>`.
 
 * **Are autoscaling Kubernetes clusters supported?**
 
diff --git a/examples/airflow/README.md b/examples/airflow/README.md
index 34390da61..b2bc19484 100644
--- a/examples/airflow/README.md
+++ b/examples/airflow/README.md
@@ -86,13 +86,13 @@ Note: `eval` can be optionally run on the same cluster as `train` with `sky exec
 
 Once we have developed the tasks, we can seamlessly run them in Airflow.
 
-1. **No changes required to our tasks -** we use the same YAMLs we wrote in the previous step to create an Airflow DAG in `sky_k8s_train_pipeline.py`.
+1. **No changes required to our tasks -** we use the same YAMLs we wrote in the previous step to create an Airflow DAG in `sky_train_dag.py`.
 2. **Airflow native logging** - SkyPilot logs are written to container stdout, which is captured as task logs in Airflow and displayed in the UI.
 3. **Easy debugging** - If a task fails, you can independently run the task using `sky launch` to debug the issue. SkyPilot will recreate the environment in which the task failed. 
 
-Here's a snippet of the DAG declaration in [sky_k8s_train_pipeline.py](https://github.com/skypilot-org/skypilot/blob/master/examples/airflow/sky_k8s_train_pipeline.py):
+Here's a snippet of the DAG declaration in [sky_train_dag.py](https://github.com/skypilot-org/skypilot/blob/master/examples/airflow/sky_train_dag.py):
 ```python
-with DAG(dag_id='sky_k8s_train_pipeline',
+with DAG(dag_id='sky_train_dag',
          default_args=default_args,
          schedule_interval=None,
          catchup=False) as dag:
@@ -143,12 +143,12 @@ All clusters are set to auto-down after the task is done, so no dangling cluster
 
 1. Copy the DAG file to the Airflow DAGs directory.
    ```bash
-   cp sky_k8s_train_pipeline.py /path/to/airflow/dags                                               
+   cp sky_train_dag.py /path/to/airflow/dags                                               
    # If your Airflow is running on Kubernetes, you may use kubectl cp to copy the file to the pod
-   # kubectl cp sky_k8s_example.py <airflow-pod-name>:/opt/airflow/dags 
+   # kubectl cp sky_train_dag.py <airflow-pod-name>:/opt/airflow/dags 
    ```
 2. Run `airflow dags list` to confirm that the DAG is loaded. 
-3. Find the DAG in the Airflow UI (typically http://localhost:8080) and enable it. The UI may take a couple of minutes to reflect the changes. Force unpause the DAG if it is paused with `airflow dags unpause sky_k8s_train_pipeline`
+3. Find the DAG in the Airflow UI (typically http://localhost:8080) and enable it. The UI may take a couple of minutes to reflect the changes. Force unpause the DAG if it is paused with `airflow dags unpause sky_train_dag`
 4. Trigger the DAG from the Airflow UI using the `Trigger DAG` button.
 5. Navigate to the run in the Airflow UI to see the DAG progress and logs of each task.
 
diff --git a/examples/airflow/sky_k8s_train_pipeline.py b/examples/airflow/sky_train_dag.py
similarity index 99%
rename from examples/airflow/sky_k8s_train_pipeline.py
rename to examples/airflow/sky_train_dag.py
index 0602290f0..9814e8eca 100644
--- a/examples/airflow/sky_k8s_train_pipeline.py
+++ b/examples/airflow/sky_train_dag.py
@@ -129,7 +129,7 @@ def generate_bucket_uuid(**context):
     return bucket_uuid
 
 
-with DAG(dag_id='sky_k8s_train_pipeline',
+with DAG(dag_id='sky_train_dag',
          default_args=default_args,
          schedule_interval=None,
          catchup=False) as dag:
diff --git a/sky/backends/wheel_utils.py b/sky/backends/wheel_utils.py
index 805117ee2..84274007b 100644
--- a/sky/backends/wheel_utils.py
+++ b/sky/backends/wheel_utils.py
@@ -17,7 +17,7 @@ import re
 import shutil
 import subprocess
 import tempfile
-from typing import Tuple
+from typing import Optional, Tuple
 
 import filelock
 from packaging import version
@@ -149,16 +149,29 @@ def build_sky_wheel() -> Tuple[pathlib.Path, str]:
         - wheel_hash: The wheel content hash.
     """
 
-    def _get_latest_modification_time(path: pathlib.Path) -> float:
+    def _get_latest_modification_time(path: pathlib.Path) -> Optional[float]:
+        max_time = -1.
         if not path.exists():
-            return -1.
-        try:
-            return max(
-                os.path.getmtime(os.path.join(root, f))
-                for root, dirs, files in os.walk(path)
-                for f in (*dirs, *files))
-        except ValueError:
-            return -1.
+            return max_time
+        for root, dirs, files in os.walk(path):
+            # Prune __pycache__ directories to prevent walking into them and
+            # exclude them from processing
+            if '__pycache__' in dirs:
+                dirs.remove('__pycache__')
+            # Filter out .pyc files
+            filtered_files = [f for f in files if not f.endswith('.pyc')]
+            # Process remaining directories and files
+            for entry in (*dirs, *filtered_files):
+                entry_path = os.path.join(root, entry)
+                try:
+                    mtime = os.path.getmtime(entry_path)
+                    if mtime > max_time:
+                        max_time = mtime
+                except OSError:
+                    # Handle cases where file might have been deleted after
+                    # listing
+                    return None
+        return max_time
 
     # This lock prevents that the wheel is updated while being copied.
     # Although the current caller already uses a lock, we still lock it here
@@ -170,10 +183,12 @@ def build_sky_wheel() -> Tuple[pathlib.Path, str]:
         last_modification_time = _get_latest_modification_time(SKY_PACKAGE_PATH)
         last_wheel_modification_time = _get_latest_modification_time(WHEEL_DIR)
 
-        # Only build wheels if the wheel is outdated or wheel does not exist
-        # for the requested version.
-        if (last_wheel_modification_time < last_modification_time) or not any(
-                WHEEL_DIR.glob(f'**/{_WHEEL_PATTERN}')):
+        # Only build wheels if the wheel is outdated, wheel does not exist
+        # for the requested version, or files were deleted during checking.
+        if ((last_modification_time is None or
+             last_wheel_modification_time is None) or
+            (last_wheel_modification_time < last_modification_time) or
+                not any(WHEEL_DIR.glob(f'**/{_WHEEL_PATTERN}'))):
             if not WHEEL_DIR.exists():
                 WHEEL_DIR.mkdir(parents=True, exist_ok=True)
             latest_wheel = _build_sky_wheel()
diff --git a/sky/cli.py b/sky/cli.py
index 1a1d9d47c..8d811cb17 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -1874,7 +1874,7 @@ def status(verbose: bool, refresh: bool, ip: bool, endpoints: bool,
                 num_in_progress_jobs, msg = _handle_jobs_queue_request(
                     managed_jobs_queue_request_id,
                     show_all=False,
-                    show_user=False,
+                    show_user=all_users,
                     limit_num_jobs_to_show=not all,
                     is_called_by_user=False)
             except KeyboardInterrupt:
diff --git a/sky/clouds/service_catalog/data_fetchers/fetch_vast.py b/sky/clouds/service_catalog/data_fetchers/fetch_vast.py
index 1426d1915..9fb6789df 100755
--- a/sky/clouds/service_catalog/data_fetchers/fetch_vast.py
+++ b/sky/clouds/service_catalog/data_fetchers/fetch_vast.py
@@ -9,8 +9,8 @@ import collections
 import csv
 import json
 import math
+import os
 import re
-import sys
 from typing import Any, Dict, List
 
 from sky.adaptors import vast
@@ -38,6 +38,10 @@ def dot_get(d: dict, key: str) -> Any:
 
 if __name__ == '__main__':
     seen = set()
+    # InstanceList is the buffered list to emit to
+    # the CSV
+    csvList = []
+
     # InstanceType and gpuInfo are basically just stubs
     # so that the dictwriter is happy without weird
     # code.
@@ -47,8 +51,6 @@ if __name__ == '__main__':
                    ('cpu_ram', 'MemoryGiB'), ('gpu_name', 'GpuInfo'),
                    ('search.totalHour', 'Price'), ('min_bid', 'SpotPrice'),
                    ('geolocation', 'Region'))
-    writer = csv.DictWriter(sys.stdout, fieldnames=[x[1] for x in mapped_keys])
-    writer.writeheader()
 
     # Vast has a wide variety of machines, some of
     # which will have less diskspace and network
@@ -141,7 +143,15 @@ if __name__ == '__main__':
                 printstub = f'{stub}#print'
                 if printstub not in seen:
                     instance['SpotPrice'] = f'{maxBid:.2f}'
-                    writer.writerow(instance)
+                    csvList.append(instance)
                     seen.add(printstub)
             else:
                 seen.add(stub)
+
+    os.makedirs('vast', exist_ok=True)
+    with open('vast/vms.csv', 'w', newline='', encoding='utf-8') as csvfile:
+        writer = csv.DictWriter(csvfile, fieldnames=[x[1] for x in mapped_keys])
+        writer.writeheader()
+
+        for instance in csvList:
+            writer.writerow(instance)
diff --git a/sky/jobs/server/core.py b/sky/jobs/server/core.py
index 78d274d3a..6a0666b8f 100644
--- a/sky/jobs/server/core.py
+++ b/sky/jobs/server/core.py
@@ -157,6 +157,7 @@ def launch(
             'modified_catalogs':
                 service_catalog_common.get_modified_catalog_file_mounts(),
             'dashboard_setup_cmd': managed_job_constants.DASHBOARD_SETUP_CMD,
+            'dashboard_user_id': common.SERVER_ID,
             **controller_utils.shared_controller_vars_to_fill(
                 controller_utils.Controllers.JOBS_CONTROLLER,
                 remote_user_config_path=remote_user_config_path,
@@ -305,10 +306,9 @@ def _maybe_restart_controller(
     with rich_utils.safe_status(
             ux_utils.spinner_message('Starting dashboard...')):
         runner = handle.get_command_runners()[0]
-        user_hash = common_utils.get_user_hash()
         runner.run(
             f'export '
-            f'{skylet_constants.USER_ID_ENV_VAR}={user_hash!r}; '
+            f'{skylet_constants.USER_ID_ENV_VAR}={common.SERVER_ID!r}; '
             f'{managed_job_constants.DASHBOARD_SETUP_CMD}',
             stream_logs=True,
         )
@@ -341,6 +341,8 @@ def queue(refresh: bool,
                 'status': (sky.jobs.ManagedJobStatus) of the job,
                 'cluster_resources': (str) resources of the cluster,
                 'region': (str) region of the cluster,
+                'user_name': (Optional[str]) job creator's user name,
+                'user_hash': (str) job creator's user hash,
             }
         ]
     Raises:
diff --git a/sky/jobs/server/server.py b/sky/jobs/server/server.py
index b95ca66d4..18d309b4c 100644
--- a/sky/jobs/server/server.py
+++ b/sky/jobs/server/server.py
@@ -159,6 +159,9 @@ async def dashboard(request: fastapi.Request,
             except Exception as e:  # pylint: disable=broad-except
                 # We catch all exceptions to gracefully handle unknown
                 # errors and retry or raise an HTTPException to the client.
+                # Assume an exception indicates that the dashboard connection
+                # is stale - remove it so that a new one is created.
+                dashboard_utils.remove_dashboard_session(user_hash)
                 msg = (
                     f'Dashboard connection attempt {attempt + 1} failed with '
                     f'{common_utils.format_exception(e, use_bracket=True)}')
diff --git a/sky/jobs/utils.py b/sky/jobs/utils.py
index 2585a676a..b73244bd4 100644
--- a/sky/jobs/utils.py
+++ b/sky/jobs/utils.py
@@ -735,6 +735,10 @@ def stream_logs_by_id(job_id: int, follow: bool = True) -> Tuple[str, int]:
         managed_job_status = managed_job_state.get_status(job_id)
         assert managed_job_status is not None, job_id
 
+    if not follow and not managed_job_status.is_terminal():
+        # The job is not in terminal state and we are not following,
+        # just return.
+        return '', exceptions.JobExitCode.SUCCEEDED
     logger.info(
         ux_utils.finishing_message(f'Managed job finished: {job_id} '
                                    f'(status: {managed_job_status.value}).'))
@@ -914,6 +918,10 @@ def load_managed_job_queue(payload: str) -> List[Dict[str, Any]]:
     jobs = message_utils.decode_payload(payload)
     for job in jobs:
         job['status'] = managed_job_state.ManagedJobStatus(job['status'])
+        if 'user_hash' in job and job['user_hash'] is not None:
+            # Skip jobs that do not have user_hash info.
+            # TODO(cooperc): Remove check before 0.12.0.
+            job['user_name'] = global_user_state.get_user(job['user_hash']).name
     return jobs
 
 
@@ -1043,16 +1051,21 @@ def format_job_table(
     def get_user_column_values(task: Dict[str, Any]) -> List[str]:
         user_values: List[str] = []
         if show_user:
+            user_name = '-'  # default value
+
+            task_user_name = task.get('user_name', None)
+            task_user_hash = task.get('user_hash', None)
+            if task_user_name is not None:
+                user_name = task_user_name
+            elif task_user_hash is not None:
+                # Fallback to the user hash if we are somehow missing the name.
+                user_name = task_user_hash
 
-            user_name = '-'
-            user_hash = task.get('user_hash', None)
-            if user_hash:
-                user = global_user_state.get_user(user_hash)
-                user_name = user.name if user.name else '-'
             user_values = [user_name]
 
             if show_all:
-                user_values.append(user_hash if user_hash is not None else '-')
+                user_values.append(
+                    task_user_hash if task_user_hash is not None else '-')
 
         return user_values
 
diff --git a/sky/server/server.py b/sky/server/server.py
index acad87f6e..e8e392c76 100644
--- a/sky/server/server.py
+++ b/sky/server/server.py
@@ -45,6 +45,7 @@ from sky.utils import admin_policy_utils
 from sky.utils import common as common_lib
 from sky.utils import common_utils
 from sky.utils import dag_utils
+from sky.utils import env_options
 from sky.utils import status_lib
 
 # pylint: disable=ungrouped-imports
@@ -65,6 +66,10 @@ def _add_timestamp_prefix_for_server_logs() -> None:
     # Add date prefix to the log message printed by loggers under
     # server.
     stream_handler = logging.StreamHandler(sys.stdout)
+    if env_options.Options.SHOW_DEBUG_INFO.get():
+        stream_handler.setLevel(logging.DEBUG)
+    else:
+        stream_handler.setLevel(logging.INFO)
     stream_handler.flush = sys.stdout.flush  # type: ignore
     stream_handler.setFormatter(sky_logging.FORMATTER)
     server_logger.addHandler(stream_handler)
diff --git a/sky/templates/jobs-controller.yaml.j2 b/sky/templates/jobs-controller.yaml.j2
index 7f78a397c..4f154ba6d 100644
--- a/sky/templates/jobs-controller.yaml.j2
+++ b/sky/templates/jobs-controller.yaml.j2
@@ -39,9 +39,7 @@ setup: |
   After=network.target
 
   [Service]
-  Environment="PATH={{ sky_python_env_path }}:\$PATH"
-  Environment="SKYPILOT_USER_ID={{controller_envs.SKYPILOT_USER_ID}}"
-  Environment="SKYPILOT_USER={{controller_envs.SKYPILOT_USER}}"
+  Environment="SKYPILOT_USER_ID={{ dashboard_user_id }}"
   Restart=always
   StandardOutput=append:/home/$USER/.sky/job-dashboard.log
   StandardError=append:/home/$USER/.sky/job-dashboard.log
@@ -51,6 +49,7 @@ setup: |
   WantedBy=default.target
   EOF
 
+  export SKYPILOT_USER_ID="{{ dashboard_user_id }}"
   {{ dashboard_setup_cmd }}
 
 run: |
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index b30f4a522..be2e710ce 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -531,8 +531,8 @@ class SSHCommandRunner(CommandRunner):
         if port_forward is not None:
             for local, remote in port_forward:
                 logger.info(
-                    f'Forwarding port {local} to port {remote} on localhost.')
-                ssh += ['-NL', f'{remote}:localhost:{local}']
+                    f'Forwarding local port {local} to remote port {remote}.')
+                ssh += ['-NL', f'{local}:localhost:{remote}']
         if self._docker_ssh_proxy_command is not None:
             docker_ssh_proxy_command = self._docker_ssh_proxy_command(ssh)
         else:
diff --git a/sky/utils/rich_utils.py b/sky/utils/rich_utils.py
index 54e9672f0..2f4dae300 100644
--- a/sky/utils/rich_utils.py
+++ b/sky/utils/rich_utils.py
@@ -230,15 +230,53 @@ def decode_rich_status(
     decoding_status = None
     try:
         last_line = ''
+        # Buffer to store incomplete UTF-8 bytes between chunks
+        undecoded_buffer = b''
+
         # Iterate over the response content in chunks. We do not use iter_lines
         # because it will strip the trailing newline characters, causing the
         # progress bar ending with `\r` becomes a pyramid.
-        for encoded_msg in response.iter_content(chunk_size=None):
-            if encoded_msg is None:
+        for chunk in response.iter_content(chunk_size=None):
+            if chunk is None:
                 return
-            encoded_msg = encoded_msg.decode('utf-8')
+
+            # Append the new chunk to any leftover bytes from previous iteration
+            current_bytes = undecoded_buffer + chunk
+            undecoded_buffer = b''
+
+            # Try to decode the combined bytes
+            try:
+                encoded_msg = current_bytes.decode('utf-8')
+            except UnicodeDecodeError as e:
+                # Check if this is potentially an incomplete sequence at the end
+                if e.start > 0:
+                    # Decode the valid part
+                    encoded_msg = current_bytes[:e.start].decode('utf-8')
+
+                    # Check if the remaining bytes are likely a partial char
+                    # or actually invalid UTF-8
+                    remaining_bytes = current_bytes[e.start:]
+                    if len(remaining_bytes) < 4:  # Max UTF-8 char is 4 bytes
+                        # Likely incomplete - save for next chunk
+                        undecoded_buffer = remaining_bytes
+                    else:
+                        # Likely invalid - replace with replacement character
+                        encoded_msg += remaining_bytes.decode('utf-8',
+                                                              errors='replace')
+                        undecoded_buffer = b''
+                else:
+                    # Error at the very beginning of the buffer - invalid UTF-8
+                    encoded_msg = current_bytes.decode('utf-8',
+                                                       errors='replace')
+                    undecoded_buffer = b''
+
             lines = encoded_msg.splitlines(keepends=True)
 
+            # Skip processing if lines is empty to avoid IndexError
+            if not lines:
+                continue
+
+            # Append any leftover text from previous chunk to first line
             lines[0] = last_line + lines[0]
             last_line = lines[-1]
             # If the last line is not ended with `\r` or `\n` (with ending


Test patch:
diff --git a/tests/unit_tests/sky/utils/test_rich_utils.py b/tests/unit_tests/sky/utils/test_rich_utils.py
new file mode 100644
index 000000000..36239cb3c
--- /dev/null
+++ b/tests/unit_tests/sky/utils/test_rich_utils.py
@@ -0,0 +1,83 @@
+import unittest
+from unittest import mock
+
+import requests
+
+from sky.utils import rich_utils
+
+
+class RichUtilsTest(unittest.TestCase):
+
+    def test_decode_rich_status_with_split_utf8(self):
+        """Test that decode_rich_status handles UTF-8 characters split across chunks."""
+        # This is a progress bar with UTF-8 box drawing characters
+        # The character █ is a multibyte UTF-8 character (0xE2 0x96 0x88)
+        progress_bar = "20%|████        | 115/581 [00:12<00:41, 11.20it/s]\n"
+
+        # Create a response object with a custom content iterator
+        mock_response = mock.MagicMock(spec=requests.Response)
+
+        # Convert to bytes and deliberately split in the middle of a multibyte character
+        # The character █ is encoded as three bytes: \xe2\x96\x88
+        progress_bytes = progress_bar.encode('utf-8')
+
+        # Find position of the first █ character and split between its bytes
+        pos = progress_bytes.find(b'\xe2\x96\x88')
+
+        # Split the bytes so that \xe2 is in one chunk and \x96\x88 is in the next
+        chunk1 = progress_bytes[:pos + 1]  # Include the first byte of █ (\xe2)
+        chunk2 = progress_bytes[pos +
+                                1:]  # Rest of the bytes including \x96\x88
+
+        # Configure the mock to return these chunks when iterated
+        mock_response.iter_content.return_value = [chunk1, chunk2]
+
+        # The function should handle the split character correctly
+        result = list(rich_utils.decode_rich_status(mock_response))
+
+        # Verify the output contains the correctly decoded progress bar
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0], progress_bar)
+
+    def test_decode_rich_status_with_multiple_split_characters(self):
+        """Test handling multiple UTF-8 characters split across several chunks."""
+        # Progress bar with multiple multibyte characters
+        progress_bar = "50%|█████████████████████████████░░░░░░░░░░░░░| 250/500 [01:15<01:15, 3.33it/s]\n"
+        progress_bytes = progress_bar.encode('utf-8')
+
+        # Create chunks that split at several different points
+        # Find positions of several █ characters and split between their bytes
+        multibyte_char = '█'.encode('utf-8')  # \xe2\x96\x88
+        positions = []
+        start = 0
+        while True:
+            pos = progress_bytes.find(multibyte_char, start)
+            if pos == -1 or len(positions) >= 3:
+                break
+            positions.append(pos)
+            start = pos + len(multibyte_char)
+
+        # Create chunks that split at different multibyte character boundaries
+        chunks = []
+        last_pos = 0
+        for i, pos in enumerate(positions):
+            if i % 2 == 0:  # For even indices, split after first byte of multibyte char
+                chunks.append(progress_bytes[last_pos:pos + 1])
+                last_pos = pos + 1
+            else:  # For odd indices, split normally
+                chunks.append(progress_bytes[last_pos:pos])
+                last_pos = pos
+
+        # Add the final chunk
+        chunks.append(progress_bytes[last_pos:])
+
+        # Configure mock response
+        mock_response = mock.MagicMock(spec=requests.Response)
+        mock_response.iter_content.return_value = chunks
+
+        # Should handle multiple split characters correctly
+        result = list(rich_utils.decode_rich_status(mock_response))
+
+        # Verify the output
+        joined_result = ''.join(r for r in result if r is not None)
+        self.assertEqual(joined_result, progress_bar)