Repository: skypilot-org/skypilot
Pull request #5918 (2025-06-07 02:19:37)

[API server] add extraInitContainers field to helm deployment
<!-- Describe the changes in this PR -->
Allowing user-specified init containers to be deployed alongside the API server container in a sidecar pattern can be helpful. One example is running Cloud SQL Auth Proxy as a sidecar to facilitate GKE pod access to GCP Cloud SQL server.

<!-- 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):

- [ ] Code formatting: install pre-commit (auto-check on commit) or `bash format.sh`
- [ ] Any manual or new tests for this PR (please specify below)
- [ ] All smoke tests: `/smoke-test` (CI) or `pytest tests/test_smoke.py` (local)
- [ ] Relevant individual tests: `/smoke-test -k test_name` (CI) or `pytest tests/test_smoke.py::test_name` (local)
- [ ] Backward compatibility: `/quicktest-core` (CI) or `pytest tests/smoke_tests/test_backward_compat.py` (local)

<!-- CI commands (/-prefixed) can only be triggered by repo members -->



Patch:
diff --git a/.dockerignore b/.dockerignore
index f3b64113e..832d0a388 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -1,1 +1,4 @@
 **/.git
+**/__pycache__
+sky/dashboard/node_modules
+sky/dashboard/.next
diff --git a/charts/skypilot/templates/api-deployment.yaml b/charts/skypilot/templates/api-deployment.yaml
index cbe495f45..4ea457585 100644
--- a/charts/skypilot/templates/api-deployment.yaml
+++ b/charts/skypilot/templates/api-deployment.yaml
@@ -323,6 +323,10 @@ spec:
         - name: lambda-config
           mountPath: /root/.lambda_cloud
       {{- end }}
+      {{- if .Values.extraInitContainers }}
+      {{- toYaml .Values.extraInitContainers | nindent 6 }}
+      {{- end }}
+
       volumes:
       {{- if .Values.storage.enabled }}
       - name: state-volume
diff --git a/charts/skypilot/values.yaml b/charts/skypilot/values.yaml
index b2c50b02a..c2e49dbad 100644
--- a/charts/skypilot/values.yaml
+++ b/charts/skypilot/values.yaml
@@ -281,6 +281,9 @@ nebiusCredentials:
   # Name of the secret containing the Nebius credentials. Only used if enabled is true.
   nebiusSecretName: nebius-credentials
 
+# Extra init containers to run before the api container
+extraInitContainers:
+
 # Set securityContext for the api pod
 podSecurityContext: {}
 
diff --git a/docs/source/admin/workspaces.rst b/docs/source/admin/workspaces.rst
index d40f647aa..e622416c7 100644
--- a/docs/source/admin/workspaces.rst
+++ b/docs/source/admin/workspaces.rst
@@ -44,6 +44,12 @@ The above is achieved by configuring the following section in the config file:
        aws:
          disabled: true
 
+      team-c:
+         private: true
+         allowed_users:
+         - alice@example.com
+         - bob@example.com
+
 
 .. dropdown:: Schema of the ``workspaces`` field:
 
@@ -192,6 +198,33 @@ workspace they belong to. Users can filter the view by workspace. Example:
 .. image:: ../images/workspaces/resources.png
    :alt: SkyPilot dashboard workspaces resources
 
+
+Private workspaces
+------------------
+
+
+Private workspaces are workspaces that are only accessible to a specific set of users.
+
+To create a private workspace, set the ``private`` field to ``true`` in the workspace configuration.
+
+.. code-block:: yaml
+
+   workspaces:
+     private-ws:
+       private: true
+       allowed_users:
+         - alice@example.com
+         - bob@example.com
+
+
+Private workspaces can only be accessed by admins or the users in the ``allowed_users`` list. For
+users who cannot access the workspace, they cannot see/access/operate on the workspace or the resources in the workspace.
+
+.. note::
+
+   The ``allowed_users`` field can be a list of user names or IDs. Note, if you
+   have multiple users with the same name, you need to specify the user IDs instead.
+
 User management
 ----------------
 
diff --git a/docs/source/reference/config.rst b/docs/source/reference/config.rst
index 472747abb..a4a181a91 100644
--- a/docs/source/reference/config.rst
+++ b/docs/source/reference/config.rst
@@ -1246,6 +1246,8 @@ Default role for users (optional).  Either ``admin`` or ``user``.
 
 If not specified, the default role is ``admin``.
 
+Note: RBAC is only functional when :ref:`Auth Proxy <api-server-auth-proxy>` is configured.
+
 .. _config-yaml-db:
 
 ``db``
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index a4cc9a444..2bd3a269d 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -51,6 +51,7 @@ from sky.utils import status_lib
 from sky.utils import subprocess_utils
 from sky.utils import timeline
 from sky.utils import ux_utils
+from sky.workspaces import core as workspaces_core
 
 if typing.TYPE_CHECKING:
     import requests
@@ -2570,19 +2571,34 @@ def get_clusters(
             public clouds only, and 'local' for only local clouds.
         cluster_names: If provided, only return records for the given cluster
             names.
+        all_users: If True, return clusters from all users. If False, only
+            return clusters from the current user.
 
     Returns:
         A list of cluster records. If the cluster does not exist or has been
         terminated, the record will be omitted from the returned list.
     """
     records = global_user_state.get_clusters()
+    current_user = common_utils.get_current_user()
+
+    # Filter by user if requested
     if not all_users:
-        current_user_hash = common_utils.get_user_hash()
         records = [
             record for record in records
-            if record['user_hash'] == current_user_hash
+            if record['user_hash'] == current_user.id
         ]
 
+    accessible_workspaces = workspaces_core.get_workspaces()
+
+    workspace_filtered_records = []
+    for record in records:
+        cluster_workspace = record.get('workspace',
+                                       constants.SKYPILOT_DEFAULT_WORKSPACE)
+        if cluster_workspace in accessible_workspaces:
+            workspace_filtered_records.append(record)
+
+    records = workspace_filtered_records
+
     yellow = colorama.Fore.YELLOW
     bright = colorama.Style.BRIGHT
     reset = colorama.Style.RESET_ALL
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index bbd93644b..51ae83625 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -2904,14 +2904,6 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
                 # TODO(suquark): once we have sky on PyPI, we should directly
                 # install sky from PyPI.
                 local_wheel_path, wheel_hash = wheel_utils.build_sky_wheel()
-                # The most frequent reason for the failure of a provision
-                # request is resource unavailability instead of rate
-                # limiting; to make users wait shorter, we do not make
-                # backoffs exponential.
-                backoff = common_utils.Backoff(
-                    initial_backoff=_RETRY_UNTIL_UP_INIT_GAP_SECONDS,
-                    max_backoff_factor=1)
-            attempt_cnt = 1
             while True:
                 # For on-demand instances, RetryingVmProvisioner will retry
                 # within the given region first, then optionally retry on all
@@ -2955,19 +2947,16 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
                         error_message = str(e)
 
                     if retry_until_up:
-                        logger.error(error_message)
-                        # Sleep and retry.
-                        gap_seconds = backoff.current_backoff()
-                        plural = 's' if attempt_cnt > 1 else ''
+                        gap_seconds = _RETRY_UNTIL_UP_INIT_GAP_SECONDS
                         retry_message = ux_utils.retry_message(
-                            f'Retry after {gap_seconds:.0f}s '
-                            f'({attempt_cnt} attempt{plural}). ')
-                        logger.info(f'\n{retry_message} '
-                                    f'{ux_utils.log_path_hint(log_path)}'
-                                    f'{colorama.Style.RESET_ALL}')
-                        attempt_cnt += 1
-                        time.sleep(gap_seconds)
-                        continue
+                            f'Retry after {gap_seconds:.0f}s ')
+                        hint_message = (f'\n{retry_message} '
+                                        f'{ux_utils.log_path_hint(log_path)}'
+                                        f'{colorama.Style.RESET_ALL}')
+                        raise exceptions.ExecutionRetryableError(
+                            error_message,
+                            hint=hint_message,
+                            retry_wait_seconds=gap_seconds)
                     # Clean up the cluster's entry in `sky status`.
                     # Do not remove the stopped cluster from the global state
                     # if failed to start.
diff --git a/sky/check.py b/sky/check.py
index 95df32c3e..23684ed66 100644
--- a/sky/check.py
+++ b/sky/check.py
@@ -1,6 +1,5 @@
 """Credential checks: check cloud credentials and enable clouds."""
 import collections
-import itertools
 import os
 import traceback
 from types import ModuleType
@@ -70,14 +69,17 @@ def check_capabilities(
 
         def check_one_cloud_one_capability(
             payload: Tuple[Tuple[str, Union[sky_clouds.Cloud, ModuleType]],
-                           sky_cloud.CloudCapability]
+                           sky_cloud.CloudCapability, bool]
         ) -> Optional[Tuple[sky_cloud.CloudCapability, bool, Optional[Union[
                 str, Dict[str, str]]]]]:
+            cloud_tuple, capability, allowed = payload
+            if not allowed:
+                return (capability, False, f'{cloud_tuple[0]} is not included '
+                        'in allowed_clouds in ~/.sky/config.yaml')
             with skypilot_config.local_active_workspace_ctx(
                     current_workspace_name):
                 # Have to override again for specific thread, as the
                 # local_active_workspace_ctx is thread-local.
-                cloud_tuple, capability = payload
                 _, cloud = cloud_tuple
                 try:
                     ok, reason = cloud.check_credentials(capability)
@@ -103,8 +105,10 @@ def check_capabilities(
 
         if clouds is not None:
             cloud_list = clouds
+            check_explicit = True
         else:
             cloud_list = get_all_clouds()
+            check_explicit = False
 
         clouds_to_check = [get_cloud_tuple(c) for c in cloud_list]
 
@@ -135,24 +139,15 @@ def check_capabilities(
         disallowed_cloud_names = [
             c for c in get_all_clouds() if c not in config_allowed_cloud_names
         ]
-        # Check only the clouds which are allowed in the config.
-        clouds_to_check = [
-            c for c in clouds_to_check if c[0] in config_allowed_cloud_names
-        ]
 
-        combinations = list(itertools.product(clouds_to_check, capabilities))
+        combinations = []
+        for c in clouds_to_check:
+            allowed = c[0] in config_allowed_cloud_names
+            if allowed or check_explicit:
+                for capability in capabilities:
+                    combinations.append((c, capability, allowed))
 
         cloud2ctx2text: Dict[str, Dict[str, str]] = {}
-        if not config_allowed_cloud_names:
-            for capability in capabilities:
-                global_user_state.set_enabled_clouds([], capability,
-                                                     current_workspace_name)
-        if not combinations:
-            echo(
-                _summary_message(enabled_clouds, cloud2ctx2text,
-                                 current_workspace_name, hide_workspace_str,
-                                 disallowed_cloud_names))
-            return {}
 
         workspace_str = f' for workspace: {current_workspace_name!r}'
         if hide_workspace_str:
@@ -172,7 +167,7 @@ def check_capabilities(
             if check_result is None:
                 continue
             capability, ok, ctx2text = check_result
-            cloud_tuple, _ = combination
+            cloud_tuple, _, _ = combination
             cloud_repr = cloud_tuple[0]
             if isinstance(ctx2text, dict):
                 cloud2ctx2text[cloud_repr] = ctx2text
@@ -345,18 +340,19 @@ def get_cached_enabled_clouds_or_refresh(
         exceptions.NoCloudAccessError: if no public cloud is enabled and
             raise_if_no_cloud_access is set to True.
     """
+    active_workspace = skypilot_config.get_active_workspace()
     cached_enabled_clouds = global_user_state.get_cached_enabled_clouds(
-        capability, skypilot_config.get_active_workspace())
+        capability, active_workspace)
     if not cached_enabled_clouds:
         try:
-            check_capability(capability, quiet=True)
+            check_capability(capability, quiet=True, workspace=active_workspace)
         except SystemExit:
             # If no cloud is enabled, check() will raise SystemExit.
             # Here we catch it and raise the exception later only if
             # raise_if_no_cloud_access is set to True.
             pass
         cached_enabled_clouds = global_user_state.get_cached_enabled_clouds(
-            capability, skypilot_config.get_active_workspace())
+            capability, active_workspace)
     if raise_if_no_cloud_access and not cached_enabled_clouds:
         with ux_utils.print_exception_no_traceback():
             raise exceptions.NoCloudAccessError(
diff --git a/sky/cli.py b/sky/cli.py
index f0883640f..f7d034d2c 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -28,7 +28,6 @@ import copy
 import datetime
 import fnmatch
 import functools
-import getpass
 import os
 import pathlib
 import shlex
@@ -6243,18 +6242,18 @@ def api_info():
     """Shows the SkyPilot API server URL."""
     url = server_common.get_server_url()
     api_server_info = sdk.api_info()
-    user_name = os.getenv(constants.USER_ENV_VAR, getpass.getuser())
-    user_hash = common_utils.get_user_hash()
     api_server_user = api_server_info.get('user')
     if api_server_user is not None:
-        user_name = api_server_user['name']
-        user_hash = api_server_user['id']
+        user = models.User(id=api_server_user['id'],
+                           name=api_server_user['name'])
+    else:
+        user = models.User.get_current_user()
     dashboard_url = server_common.get_dashboard_url(url)
     click.echo(f'Using SkyPilot API server: {url}\n'
                f'{ux_utils.INDENT_SYMBOL}Status: {api_server_info["status"]}, '
                f'commit: {api_server_info["commit"]}, '
                f'version: {api_server_info["version"]}\n'
-               f'{ux_utils.INDENT_SYMBOL}User: {user_name} ({user_hash})\n'
+               f'{ux_utils.INDENT_SYMBOL}User: {user.name} ({user.id})\n'
                f'{ux_utils.INDENT_LAST_SYMBOL}Dashboard: {dashboard_url}')
 
 
diff --git a/sky/core.py b/sky/core.py
index 8ee72c90d..115f41f82 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -711,9 +711,10 @@ def queue(cluster_name: str,
         exceptions.CommandError: if failed to get the job queue with ssh.
     """
     all_jobs = not skip_finished
-    user_hash: Optional[str] = common_utils.get_user_hash()
     if all_users:
         user_hash = None
+    else:
+        user_hash = common_utils.get_current_user().id
     code = job_lib.JobLibCodeGen.get_job_queue(user_hash, all_jobs)
 
     handle = backend_utils.check_cluster_available(
@@ -829,7 +830,7 @@ def cancel(
     backend.cancel_jobs(handle,
                         job_ids,
                         cancel_all=all or all_users,
-                        user_hash=common_utils.get_user_hash())
+                        user_hash=common_utils.get_current_user().id)
 
 
 @usage_lib.entrypoint
diff --git a/sky/dashboard/src/components/elements/ErrorDisplay.jsx b/sky/dashboard/src/components/elements/ErrorDisplay.jsx
index 3e9e78413..6318b3faa 100644
--- a/sky/dashboard/src/components/elements/ErrorDisplay.jsx
+++ b/sky/dashboard/src/components/elements/ErrorDisplay.jsx
@@ -11,8 +11,7 @@ const cleanErrorMessage = (error) => {
     message = message.split('failed:')[1].trim();
   }
 
-  // Capitalize first letter and return
-  return message.charAt(0).toUpperCase() + message.slice(1);
+  return message;
 };
 
 // Error display component
diff --git a/sky/dashboard/src/components/elements/modals.jsx b/sky/dashboard/src/components/elements/modals.jsx
index 3112b7603..9fb916463 100644
--- a/sky/dashboard/src/components/elements/modals.jsx
+++ b/sky/dashboard/src/components/elements/modals.jsx
@@ -178,6 +178,7 @@ export function ConfirmationModal({
   message,
   confirmText = 'Confirm',
   confirmVariant = 'destructive',
+  confirmClassName = null,
 }) {
   return (
     <Dialog open={isOpen} onOpenChange={onClose}>
@@ -191,7 +192,8 @@ export function ConfirmationModal({
             Cancel
           </Button>
           <Button
-            variant={confirmVariant}
+            variant={confirmClassName ? undefined : confirmVariant}
+            className={confirmClassName}
             onClick={() => {
               onConfirm();
               onClose();
diff --git a/sky/dashboard/src/components/elements/sidebar.jsx b/sky/dashboard/src/components/elements/sidebar.jsx
index 816188d44..3bac0b79b 100755
--- a/sky/dashboard/src/components/elements/sidebar.jsx
+++ b/sky/dashboard/src/components/elements/sidebar.jsx
@@ -20,8 +20,9 @@ import {
   BookDocIcon,
   UserCircleIcon,
   UsersIcon,
+  StarIcon,
 } from '@/components/elements/icons';
-import { Settings } from 'lucide-react';
+import { Settings, User } from 'lucide-react';
 import { BASE_PATH, ENDPOINT } from '@/data/connectors/constants';
 import { CustomTooltip } from '@/components/utils';
 import { useMobile } from '@/hooks/useMobile';
@@ -32,17 +33,40 @@ const SidebarContext = createContext(null);
 export function SidebarProvider({ children }) {
   const [isSidebarOpen, setIsSidebarOpen] = useState(true);
   const [userEmail, setUserEmail] = useState(null);
+  const [userRole, setUserRole] = useState(null);
 
   const toggleSidebar = () => {
     setIsSidebarOpen((prev) => !prev);
   };
 
   useEffect(() => {
+    // Fetch user info from health endpoint
     fetch(`${ENDPOINT}/api/health`)
       .then((res) => res.json())
       .then((data) => {
         if (data.user && data.user.name) {
           setUserEmail(data.user.name);
+
+          // Get role from direct API endpoint to avoid cache interference
+          // Using cache would cause race condition, which leads to unexpected
+          // behavior in workspaces and users page.
+          const getUserRole = async () => {
+            try {
+              const response = await fetch(`${ENDPOINT}/users/role`);
+              if (response.ok) {
+                const roleData = await response.json();
+                if (roleData.role) {
+                  setUserRole(roleData.role);
+                }
+              }
+            } catch (error) {
+              // If role data is not available or there's an error,
+              // we just don't show the role - it's not critical
+              console.log('Could not fetch user role:', error);
+            }
+          };
+
+          getUserRole();
         }
       })
       .catch((error) => {
@@ -52,7 +76,7 @@ export function SidebarProvider({ children }) {
 
   return (
     <SidebarContext.Provider
-      value={{ isSidebarOpen, toggleSidebar, userEmail }}
+      value={{ isSidebarOpen, toggleSidebar, userEmail, userRole }}
     >
       {children}
     </SidebarContext.Provider>
@@ -120,7 +144,7 @@ export function SideBar({ highlighted = 'clusters' }) {
 export function TopBar() {
   const router = useRouter();
   const isMobile = useMobile();
-  const { userEmail } = useSidebar();
+  const { userEmail, userRole } = useSidebar();
   const [isDropdownOpen, setIsDropdownOpen] = useState(false);
   const dropdownRef = useRef(null);
 
@@ -138,6 +162,19 @@ export function TopBar() {
     };
   }, [dropdownRef]);
 
+  // Function to get user initial
+  const getUserInitial = (email) => {
+    if (!email) return '?';
+
+    // If it's an email, get the first letter of the username part
+    if (email.includes('@')) {
+      return email.split('@')[0].charAt(0).toUpperCase();
+    }
+
+    // If it's just a name, get the first letter
+    return email.charAt(0).toUpperCase();
+  };
+
   // Function to determine if a path is active
   const isActivePath = (path) => {
     // Special case: highlight workspaces for both /workspaces and /workspace paths
@@ -334,12 +371,14 @@ export function TopBar() {
             <div className="relative" ref={dropdownRef}>
               <button
                 onClick={() => setIsDropdownOpen(!isDropdownOpen)}
-                className="inline-flex items-center justify-center p-2 rounded-full text-gray-600 hover:bg-gray-100 transition-colors duration-150 cursor-pointer"
+                className="inline-flex items-center justify-center rounded-full transition-colors duration-150 cursor-pointer hover:ring-2 hover:ring-blue-200"
                 title="User Profile"
               >
-                <UserCircleIcon
-                  className={`${isMobile ? 'w-5 h-5' : 'w-6 h-6'}`}
-                />
+                <div
+                  className={`${isMobile ? 'w-6 h-6' : 'w-7 h-7'} bg-blue-600 text-white rounded-full flex items-center justify-center font-medium ${isMobile ? 'text-xs' : 'text-sm'} hover:bg-blue-700 transition-colors`}
+                >
+                  {getUserInitial(userEmail)}
+                </div>
               </button>
 
               {isDropdownOpen && (
@@ -357,10 +396,25 @@ export function TopBar() {
                           {displayName}
                         </div>
                         {emailToDisplay && (
-                          <div className="px-4 pt-0 pb-2 text-xs text-gray-500">
+                          <div className="px-4 pt-0 pb-1 text-xs text-gray-500">
                             {emailToDisplay}
                           </div>
                         )}
+                        {userRole && (
+                          <div className="px-4 pt-0 pb-2 text-xs">
+                            {userRole === 'admin' ? (
+                              <span className="inline-flex items-center text-blue-600">
+                                <StarIcon className="w-3 h-3 mr-1" />
+                                Admin
+                              </span>
+                            ) : (
+                              <span className="inline-flex items-center text-gray-600">
+                                <User className="w-3 h-3 mr-1" />
+                                User
+                              </span>
+                            )}
+                          </div>
+                        )}
                       </>
                     );
                   })()}
diff --git a/sky/dashboard/src/components/jobs.jsx b/sky/dashboard/src/components/jobs.jsx
index ee04e8a53..e638096f8 100755
--- a/sky/dashboard/src/components/jobs.jsx
+++ b/sky/dashboard/src/components/jobs.jsx
@@ -26,6 +26,7 @@ import {
   CustomTooltip as Tooltip,
   NonCapitalizedTooltip,
   relativeTime,
+  formatDateTime,
 } from '@/components/utils';
 import {
   FileSearchIcon,
@@ -90,7 +91,12 @@ export function filterJobsByWorkspace(jobs, workspaceFilter) {
 const formatSubmittedTime = (timestamp) => {
   if (!timestamp) return '-';
 
-  let timeString = relativeTime(timestamp);
+  // Convert timestamp to Date object if it's not already
+  const date =
+    timestamp instanceof Date ? timestamp : new Date(timestamp * 1000);
+
+  // Get the original timeString from relativeTime
+  let timeString = relativeTime(date);
 
   // If relativeTime returns a React element, extract the string from its props
   if (
@@ -115,32 +121,72 @@ const formatSubmittedTime = (timestamp) => {
     return timeString; // Return as is if not a string after extraction
   }
 
+  // Apply the same shortening logic
+  const shortenedTime = shortenTimeForJobs(timeString);
+
+  // Return with tooltip functionality like relativeTime does
+  const now = new Date();
+  const differenceInDays =
+    (now.getTime() - date.getTime()) / (1000 * 3600 * 24);
+
+  if (Math.abs(differenceInDays) < 7) {
+    return (
+      <Tooltip
+        content={formatDateTime(date)}
+        className="capitalize text-sm text-muted-foreground"
+      >
+        {shortenedTime}
+      </Tooltip>
+    );
+  } else {
+    return (
+      <Tooltip
+        content={formatDateTime(date)}
+        className="text-sm text-muted-foreground"
+      >
+        {shortenedTime}
+      </Tooltip>
+    );
+  }
+};
+
+// Helper function to shorten time strings for jobs
+function shortenTimeForJobs(timeString) {
+  if (!timeString || typeof timeString !== 'string') {
+    return timeString;
+  }
+
   // Handle "just now" case
   if (timeString === 'just now') {
     return 'now';
   }
 
-  // Handle "about X unit(s) ago" e.g. "about 1 hour ago" -> "~1 hr ago"
+  // Handle "less than a minute ago" case
+  if (timeString.toLowerCase() === 'less than a minute ago') {
+    return 'Less than a minute ago';
+  }
+
+  // Handle "about X unit(s) ago" e.g. "about 1 hour ago" -> "1h ago"
   const aboutMatch = timeString.match(/^About\s+(\d+)\s+(\w+)\s+ago$/);
   if (aboutMatch) {
     const num = aboutMatch[1];
     const unit = aboutMatch[2];
     const unitMap = {
-      second: 'sec',
-      seconds: 'secs',
-      minute: 'min',
-      minutes: 'mins',
-      hour: 'hr',
-      hours: 'hrs',
+      second: 's',
+      seconds: 's',
+      minute: 'm',
+      minutes: 'm',
+      hour: 'h',
+      hours: 'h',
       day: 'd',
       days: 'd',
       month: 'mo',
-      months: 'mos',
+      months: 'mo',
       year: 'yr',
-      years: 'yrs',
+      years: 'yr',
     };
     if (unitMap[unit]) {
-      return `~ ${num} ${unitMap[unit]} ago`;
+      return `${num}${unitMap[unit]} ago`;
     }
   }
 
@@ -149,15 +195,15 @@ const formatSubmittedTime = (timestamp) => {
   if (singleUnitMatch) {
     const unit = singleUnitMatch[1];
     const unitMap = {
-      second: 'sec',
-      minute: 'min',
-      hour: 'hr',
+      second: 's',
+      minute: 'm',
+      hour: 'h',
       day: 'd',
       month: 'mo',
       year: 'yr',
     };
     if (unitMap[unit]) {
-      return `1 ${unitMap[unit]} ago`;
+      return `1${unitMap[unit]} ago`;
     }
   }
 
@@ -167,21 +213,27 @@ const formatSubmittedTime = (timestamp) => {
     const num = multiUnitMatch[1];
     const unit = multiUnitMatch[2];
     const unitMap = {
-      seconds: 'secs',
-      minutes: 'mins',
-      hours: 'hrs',
+      second: 's',
+      seconds: 's',
+      minute: 'm',
+      minutes: 'm',
+      hour: 'h',
+      hours: 'h',
+      day: 'd',
       days: 'd',
+      month: 'mo',
       months: 'mo',
+      year: 'yr',
       years: 'yr',
     };
     if (unitMap[unit]) {
-      return `${num} ${unitMap[unit]} ago`;
+      return `${num}${unitMap[unit]} ago`;
     }
   }
 
   // Fallback if no specific regex match (e.g., for dates beyond 7 days or other formats)
   return timeString;
-};
+}
 
 export function ManagedJobs() {
   const router = useRouter();
@@ -1006,6 +1058,7 @@ export function ManagedJobsTable({
         onConfirm={confirmationModal.onConfirm}
         title={confirmationModal.title}
         message={confirmationModal.message}
+        confirmClassName="bg-blue-600 hover:bg-blue-700 text-white"
       />
     </div>
   );
diff --git a/sky/dashboard/src/components/users.jsx b/sky/dashboard/src/components/users.jsx
index 909892c42..b3fbbeed7 100644
--- a/sky/dashboard/src/components/users.jsx
+++ b/sky/dashboard/src/components/users.jsx
@@ -33,11 +33,19 @@ import { Card } from '@/components/ui/card';
 import { ENDPOINT } from '@/data/connectors/constants';
 
 // Helper functions for username parsing
-const parseUsername = (username) => {
+const parseUsername = (username, userId) => {
   if (username && username.includes('@')) {
     return username.split('@')[0];
   }
-  return username || 'N/A';
+  // If no email, show username with userId in parentheses only if they're different
+  const usernameBase = username || 'N/A';
+
+  // Skip showing userId if it's the same as username
+  if (userId && userId !== usernameBase) {
+    return `${usernameBase} (${userId})`;
+  }
+
+  return usernameBase;
 };
 
 const getFullEmail = (username) => {
@@ -124,7 +132,7 @@ function UsersTable({ refreshInterval, setLoading, refreshDataRef }) {
         // Show users immediately with placeholder counts
         const initialProcessedUsers = (usersData || []).map((user) => ({
           ...user,
-          usernameDisplay: parseUsername(user.username),
+          usernameDisplay: parseUsername(user.username, user.userId),
           fullEmail: getFullEmail(user.username),
           clusterCount: -1, // Use -1 as loading indicator
           jobCount: -1, // Use -1 as loading indicator
@@ -155,7 +163,7 @@ function UsersTable({ refreshInterval, setLoading, refreshDataRef }) {
           );
           return {
             ...user,
-            usernameDisplay: parseUsername(user.username),
+            usernameDisplay: parseUsername(user.username, user.userId),
             fullEmail: getFullEmail(user.username),
             clusterCount: userClusters.length,
             jobCount: userJobs.length,
diff --git a/sky/dashboard/src/components/utils.jsx b/sky/dashboard/src/components/utils.jsx
index 940bcbec4..9460bf417 100644
--- a/sky/dashboard/src/components/utils.jsx
+++ b/sky/dashboard/src/components/utils.jsx
@@ -25,12 +25,15 @@ export function relativeTime(date) {
   const now = new Date();
   const differenceInDays = (now - date) / (1000 * 3600 * 24);
   if (Math.abs(differenceInDays) < 7) {
+    const originalTimeString = formatDistance(date, now, { addSuffix: true });
+    const shortenedTimeString = shortenTimeString(originalTimeString);
+
     return (
       <CustomTooltip
         content={formatDateTime(date)}
         className="capitalize text-sm text-muted-foreground"
       >
-        {capitalizeFirstWord(formatDistance(date, now, { addSuffix: true }))}
+        {shortenedTimeString}
       </CustomTooltip>
     );
   } else {
@@ -45,6 +48,86 @@ export function relativeTime(date) {
   }
 }
 
+// Helper function to shorten time strings
+function shortenTimeString(timeString) {
+  if (!timeString || typeof timeString !== 'string') {
+    return timeString;
+  }
+
+  // Handle "just now" case
+  if (timeString === 'just now') {
+    return 'now';
+  }
+
+  // Handle "about X unit(s) ago" e.g. "about 1 hour ago" -> "1h ago"
+  const aboutMatch = timeString.match(/^about\s+(\d+)\s+(\w+)\s+ago$/i);
+  if (aboutMatch) {
+    const num = aboutMatch[1];
+    const unit = aboutMatch[2];
+    const unitMap = {
+      second: 's',
+      seconds: 's',
+      minute: 'm',
+      minutes: 'm',
+      hour: 'h',
+      hours: 'h',
+      day: 'd',
+      days: 'd',
+      month: 'mo',
+      months: 'mo',
+      year: 'yr',
+      years: 'yr',
+    };
+    if (unitMap[unit]) {
+      return `${num}${unitMap[unit]} ago`;
+    }
+  }
+
+  // Handle "a minute ago", "an hour ago", etc.
+  const singleUnitMatch = timeString.match(/^a[n]?\s+(\w+)\s+ago$/i);
+  if (singleUnitMatch) {
+    const unit = singleUnitMatch[1];
+    const unitMap = {
+      second: 's',
+      minute: 'm',
+      hour: 'h',
+      day: 'd',
+      month: 'mo',
+      year: 'yr',
+    };
+    if (unitMap[unit]) {
+      return `1${unitMap[unit]} ago`;
+    }
+  }
+
+  // Handle "X units ago"
+  const multiUnitMatch = timeString.match(/^(\d+)\s+(\w+)\s+ago$/i);
+  if (multiUnitMatch) {
+    const num = multiUnitMatch[1];
+    const unit = multiUnitMatch[2];
+    const unitMap = {
+      second: 's',
+      seconds: 's',
+      minute: 'm',
+      minutes: 'm',
+      hour: 'h',
+      hours: 'h',
+      day: 'd',
+      days: 'd',
+      month: 'mo',
+      months: 'mo',
+      year: 'yr',
+      years: 'yr',
+    };
+    if (unitMap[unit]) {
+      return `${num}${unitMap[unit]} ago`;
+    }
+  }
+
+  // Fallback to original string if no patterns match
+  return timeString;
+}
+
 export function formatDateTime(date) {
   const options = {
     year: 'numeric',
diff --git a/sky/dashboard/src/components/workspace-editor.jsx b/sky/dashboard/src/components/workspace-editor.jsx
index b346425e0..85878c0c2 100644
--- a/sky/dashboard/src/components/workspace-editor.jsx
+++ b/sky/dashboard/src/components/workspace-editor.jsx
@@ -18,7 +18,13 @@ import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/card';
 import { Button } from '@/components/ui/button';
 import { Textarea } from '@/components/ui/textarea';
 import { CircularProgress } from '@mui/material';
-import { SaveIcon, TrashIcon, CheckIcon, RotateCwIcon } from 'lucide-react';
+import {
+  SaveIcon,
+  TrashIcon,
+  CheckIcon,
+  RotateCwIcon,
+  User,
+} from 'lucide-react';
 import {
   Dialog,
   DialogContent,
@@ -32,11 +38,13 @@ import {
   ServerIcon,
   BriefcaseIcon,
   TickIcon,
+  StarIcon,
 } from '@/components/elements/icons';
 import { ErrorDisplay } from '@/components/elements/ErrorDisplay';
 import { statusGroups } from './jobs'; // Import statusGroups
 import yaml from 'js-yaml';
 import { CLOUD_CONONICATIONS } from '@/data/connectors/constants';
+import { getUsers } from '@/data/connectors/users';
 
 // Success display component
 const SuccessDisplay = ({ message }) => {
@@ -79,7 +87,13 @@ const WorkspaceConfigDescription = ({
   );
 
   Object.entries(config).forEach(([cloud, cloudConfig]) => {
-    const cloudName = CLOUD_CONONICATIONS[cloud.toLowerCase()];
+    // Skip non-cloud configuration keys
+    if (cloud === 'private' || cloud === 'allowed_users') {
+      return;
+    }
+
+    const cloudName =
+      CLOUD_CONONICATIONS[cloud.toLowerCase()] || cloud.toUpperCase();
     const isActuallyEnabled = enabledCloudsSet.has(cloudName?.toLowerCase());
 
     if (cloudConfig?.disabled === true) {
@@ -162,6 +176,84 @@ const WorkspaceConfigDescription = ({
   return null;
 };
 
+// Workspace badge component for private/public status
+const WorkspaceBadge = ({ isPrivate }) => {
+  if (isPrivate) {
+    return (
+      <span className="inline-flex items-center px-2 py-1 rounded-full text-xs font-medium bg-amber-100 text-amber-800 border border-amber-300">
+        Private
+      </span>
+    );
+  }
+  return (
+    <span className="inline-flex items-center px-2 py-1 rounded-full text-xs font-medium bg-green-100 text-green-800 border border-green-300">
+      Public
+    </span>
+  );
+};
+
+// Detailed allowed users component for workspace editor
+const DetailedAllowedUsers = ({ workspaceConfig, allUsers }) => {
+  if (!workspaceConfig.private) return null;
+
+  // Get allowed users from config
+  const allowedUsersFromConfig = workspaceConfig.allowed_users || [];
+
+  // Get all admin users
+  const adminUsers = (allUsers || []).filter((user) => user.role === 'admin');
+  const adminUsernames = adminUsers.map((user) => user.username);
+
+  // Combine allowed users and admin users, remove duplicates
+  const allAllowedUsers = [
+    ...new Set([...allowedUsersFromConfig, ...adminUsernames]),
+  ];
+
+  if (allAllowedUsers.length === 0) {
+    return (
+      <div className="mt-4">
+        <h4 className="mb-2 text-xs text-gray-500 tracking-wider">
+          Allowed Users (0)
+        </h4>
+        <div className="text-amber-600 text-xs italic p-2 bg-amber-50 rounded border border-amber-200">
+          No users configured (workspace may be inaccessible)
+        </div>
+      </div>
+    );
+  }
+
+  return (
+    <div className="mt-4">
+      <h4 className="mb-2 text-xs text-gray-500 tracking-wider">
+        Allowed Users ({allAllowedUsers.length})
+      </h4>
+      <div className="space-y-1 max-h-48 overflow-y-auto border border-gray-200 rounded">
+        {allAllowedUsers.map((username) => {
+          const isAdmin = adminUsernames.includes(username);
+          return (
+            <div
+              key={username}
+              className="flex items-center justify-between text-xs p-2 bg-gray-50 hover:bg-gray-100 border-b border-gray-100 last:border-b-0"
+            >
+              <span className="font-medium text-gray-700">{username}</span>
+              {isAdmin ? (
+                <span className="inline-flex items-center text-blue-600">
+                  <StarIcon className="w-3 h-3 mr-1" />
+                  Admin
+                </span>
+              ) : (
+                <span className="inline-flex items-center text-gray-600">
+                  <User className="w-3 h-3 mr-1" />
+                  User
+                </span>
+              )}
+            </div>
+          );
+        })}
+      </div>
+    </div>
+  );
+};
+
 export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
   const router = useRouter();
   const [workspaceConfig, setWorkspaceConfig] = useState({});
@@ -173,6 +265,7 @@ export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
   const [error, setError] = useState(null);
   const [success, setSuccess] = useState(null);
   const [yamlError, setYamlError] = useState(null);
+  const [allUsers, setAllUsers] = useState([]);
 
   // Delete state
   const [deleteState, setDeleteState] = useState({
@@ -194,10 +287,15 @@ export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
     setLoading(true);
     setError(null);
     try {
-      const allWorkspaces = await getWorkspaces();
+      const [allWorkspaces, usersResponse] = await Promise.all([
+        getWorkspaces(),
+        getUsers(),
+      ]);
+
       const config = allWorkspaces[workspaceName] || {};
       setWorkspaceConfig(config);
       setOriginalConfig(config);
+      setAllUsers(usersResponse || []);
 
       // Format as YAML with workspace name as top-level key
       const fullConfig = { [workspaceName]: config };
@@ -550,8 +648,15 @@ export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
                   <Card className="h-full">
                     <CardHeader>
                       <CardTitle className="text-base font-normal">
-                        <span className="font-semibold">Workspace:</span>{' '}
-                        {workspaceName}
+                        <div className="flex items-center justify-between">
+                          <div>
+                            <span className="font-semibold">Workspace:</span>{' '}
+                            {workspaceName}
+                          </div>
+                          <WorkspaceBadge
+                            isPrivate={originalConfig.private === true}
+                          />
+                        </div>
                       </CardTitle>
                     </CardHeader>
                     <CardContent className="text-sm pb-2 flex-1">
@@ -611,6 +716,12 @@ export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
                           enabledClouds={workspaceStats.clouds}
                         />
                       </div>
+
+                      {/* Detailed allowed users for private workspaces */}
+                      <DetailedAllowedUsers
+                        workspaceConfig={originalConfig}
+                        allUsers={allUsers}
+                      />
                     </div>
                   </Card>
                 </div>
@@ -660,12 +771,16 @@ export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
                           <div className="p-3 bg-gray-50 border rounded-lg">
                             <pre className="text-xs font-mono text-gray-600 whitespace-pre-wrap">
                               {`${workspaceName || 'my-workspace'}:
+  private: true
+  allowed_users:
+  - user1@mydomain.com
+  - user2@mydomain.com
   gcp:
     project_id: xxx
     disabled: false
   kubernetes:
     allowed_contexts:
-      - context-1`}
+    - context-1`}
                             </pre>
                           </div>
                         </div>
diff --git a/sky/dashboard/src/components/workspaces.jsx b/sky/dashboard/src/components/workspaces.jsx
index 4c9faabbb..3c3603543 100644
--- a/sky/dashboard/src/components/workspaces.jsx
+++ b/sky/dashboard/src/components/workspaces.jsx
@@ -122,36 +122,52 @@ const WorkspaceConfigDescription = ({ workspaceName, config }) => {
   return null;
 };
 
-// Workspace card component
-const WorkspaceCard = ({ workspace, onDelete, onEdit, router }) => {
-  const handleEdit = async () => {
-    try {
-      // Get current user's role first
-      const response = await fetch(`${ENDPOINT}/users/role`);
-      if (!response.ok) {
-        const errorData = await response.json();
-        throw new Error(errorData.detail || 'Failed to get user role');
-      }
-      const data = await response.json();
-      const currentUserRole = data.role;
-
-      if (currentUserRole != 'admin') {
-        alert(`${data.name} is logged in as no admin, cannot edit workspace`);
-        return;
-      }
+// Workspace badge component for private/public status
+const WorkspaceBadge = ({ isPrivate }) => {
+  if (isPrivate) {
+    return (
+      <span className="inline-flex items-center px-2 py-1 rounded-full text-xs font-medium bg-amber-100 text-amber-800 border border-amber-300">
+        Private
+      </span>
+    );
+  }
+  return (
+    <span className="inline-flex items-center px-2 py-1 rounded-full text-xs font-medium bg-green-100 text-green-800 border border-green-300">
+      Public
+    </span>
+  );
+};
 
+// Workspace card component
+const WorkspaceCard = ({
+  workspace,
+  onDelete,
+  onEdit,
+  router,
+  rawWorkspacesData,
+  checkPermissionAndAct,
+  roleLoading,
+}) => {
+  const handleEdit = () => {
+    checkPermissionAndAct('cannot edit workspace', () => {
       onEdit(workspace.name);
-    } catch (error) {
-      console.error('Failed to check user role:', error);
-      alert(`Error: ${error.message}`);
-    }
+    });
   };
 
+  // Get the workspace configuration to check if it's private
+  const workspaceConfig = rawWorkspacesData?.[workspace.name] || {};
+  const isPrivate = workspaceConfig.private === true;
+
   return (
     <Card key={workspace.name}>
       <CardHeader>
         <CardTitle className="text-base font-normal">
-          <span className="font-semibold">Workspace:</span> {workspace.name}
+          <div className="flex items-center justify-between">
+            <div>
+              <span className="font-semibold">Workspace:</span> {workspace.name}
+            </div>
+            <WorkspaceBadge isPrivate={isPrivate} />
+          </div>
         </CardTitle>
       </CardHeader>
       <CardContent className="text-sm pb-2">
@@ -210,7 +226,7 @@ const WorkspaceCard = ({ workspace, onDelete, onEdit, router }) => {
           variant="outline"
           size="sm"
           onClick={() => onDelete(workspace.name)}
-          disabled={workspace.name === 'default'}
+          disabled={workspace.name === 'default' || roleLoading}
           title={
             workspace.name === 'default'
               ? 'Cannot delete default workspace'
@@ -218,10 +234,29 @@ const WorkspaceCard = ({ workspace, onDelete, onEdit, router }) => {
           }
           className="text-red-600 hover:text-red-700 hover:bg-red-50"
         >
-          Delete
+          {roleLoading ? (
+            <div className="flex items-center">
+              <CircularProgress size={12} className="mr-1" />
+              <span>Delete</span>
+            </div>
+          ) : (
+            'Delete'
+          )}
         </Button>
-        <Button variant="outline" size="sm" onClick={handleEdit}>
-          Edit
+        <Button
+          variant="outline"
+          size="sm"
+          onClick={handleEdit}
+          disabled={roleLoading}
+        >
+          {roleLoading ? (
+            <div className="flex items-center">
+              <CircularProgress size={12} className="mr-1" />
+              <span>Edit</span>
+            </div>
+          ) : (
+            'Edit'
+          )}
         </Button>
       </CardFooter>
     </Card>
@@ -229,28 +264,13 @@ const WorkspaceCard = ({ workspace, onDelete, onEdit, router }) => {
 };
 
 // Create new workspace card component
-const CreateWorkspaceCard = ({ onClick }) => {
-  const handleClick = async () => {
-    try {
-      // Get current user's role first
-      const response = await fetch(`${ENDPOINT}/users/role`);
-      if (!response.ok) {
-        const errorData = await response.json();
-        throw new Error(errorData.detail || 'Failed to get user role');
-      }
-      const data = await response.json();
-      const currentUserRole = data.role;
-
-      if (currentUserRole != 'admin') {
-        alert(`${data.name} is logged in as no admin, cannot create workspace`);
-        return;
-      }
-
-      onClick();
-    } catch (error) {
-      console.error('Failed to check user role:', error);
-      alert(`Error: ${error.message}`);
-    }
+const CreateWorkspaceCard = ({
+  onClick,
+  checkPermissionAndAct,
+  roleLoading,
+}) => {
+  const handleClick = () => {
+    checkPermissionAndAct('cannot create workspace', onClick);
   };
 
   return (
@@ -349,9 +369,76 @@ export function Workspaces() {
     error: null,
   });
 
+  // Permission denial dialog state
+  const [permissionDenialState, setPermissionDenialState] = useState({
+    open: false,
+    message: '',
+    userName: '',
+  });
+
+  // User role cache
+  const [userRoleCache, setUserRoleCache] = useState(null);
+  const [roleLoading, setRoleLoading] = useState(false);
+
   const router = useRouter();
   const isMobile = useMobile();
 
+  // Function to get user role with caching
+  const getUserRole = async () => {
+    // Return cached result if available and less than 5 minutes old
+    if (userRoleCache && Date.now() - userRoleCache.timestamp < 5 * 60 * 1000) {
+      return userRoleCache;
+    }
+
+    setRoleLoading(true);
+    try {
+      const response = await fetch(`${ENDPOINT}/users/role`);
+      if (!response.ok) {
+        const errorData = await response.json();
+        throw new Error(errorData.detail || 'Failed to get user role');
+      }
+      const data = await response.json();
+      const roleData = {
+        role: data.role,
+        name: data.name,
+        timestamp: Date.now(),
+      };
+      setUserRoleCache(roleData);
+      setRoleLoading(false);
+      return roleData;
+    } catch (error) {
+      setRoleLoading(false);
+      throw error;
+    }
+  };
+
+  // Function to handle permission check with smooth UX
+  const checkPermissionAndAct = async (action, actionCallback) => {
+    try {
+      const roleData = await getUserRole();
+
+      if (roleData.role !== 'admin') {
+        setPermissionDenialState({
+          open: true,
+          message: action,
+          userName: roleData.name.toLowerCase(),
+        });
+        return false;
+      }
+
+      actionCallback();
+      return true;
+    } catch (error) {
+      console.error('Failed to check user role:', error);
+      setPermissionDenialState({
+        open: true,
+        message: `Error: ${error.message}`,
+        userName: '',
+      });
+      return false;
+    }
+  };
+
   const fetchData = async (showLoading = false) => {
     if (showLoading) {
       setLoading(true);
@@ -489,32 +576,15 @@ export function Workspaces() {
     return () => clearInterval(interval);
   }, []);
 
-  const handleDeleteWorkspace = async (workspaceName) => {
-    try {
-      // Get current user's role first
-      const response = await fetch(`${ENDPOINT}/users/role`);
-      if (!response.ok) {
-        const errorData = await response.json();
-        throw new Error(errorData.detail || 'Failed to get user role');
-      }
-      const data = await response.json();
-      const currentUserRole = data.role;
-
-      if (currentUserRole != 'admin') {
-        alert(`${data.name} is logged in as no admin, cannot delete workspace`);
-        return;
-      }
-
+  const handleDeleteWorkspace = (workspaceName) => {
+    checkPermissionAndAct('cannot delete workspace', () => {
       setDeleteState({
         confirmOpen: true,
         workspaceToDelete: workspaceName,
         deleting: false,
         error: null,
       });
-    } catch (error) {
-      console.error('Failed to check user role:', error);
-      alert(`Error: ${error.message}`);
-    }
+    });
   };
 
   const handleConfirmDelete = async () => {
@@ -529,7 +599,11 @@ export function Workspaces() {
         deleting: false,
         error: null,
       });
-      await fetchData();
+
+      // Invalidate cache to ensure fresh data is fetched (same as manual refresh)
+      dashboardCache.invalidate(getWorkspaces);
+
+      await fetchData(true); // Show loading during refresh
     } catch (error) {
       console.error('Error deleting workspace:', error);
       setDeleteState((prev) => ({
@@ -559,27 +633,10 @@ export function Workspaces() {
     });
   };
 
-  const handleEditAllConfigs = async () => {
-    try {
-      // Get current user's role first
-      const response = await fetch(`${ENDPOINT}/users/role`);
-      if (!response.ok) {
-        const errorData = await response.json();
-        throw new Error(errorData.detail || 'Failed to get user role');
-      }
-      const data = await response.json();
-      const currentUserRole = data.role;
-
-      if (currentUserRole != 'admin') {
-        alert(`${data.name} is logged in as no admin, cannot edit config`);
-        return;
-      }
-
+  const handleEditAllConfigs = () => {
+    checkPermissionAndAct('cannot edit config', () => {
       router.push('/config');
-    } catch (error) {
-      console.error('Failed to check user role:', error);
-      alert(`Error: ${error.message}`);
-    }
+    });
   };
 
   const preStyle = {
@@ -613,11 +670,19 @@ export function Workspaces() {
             className="ml-4 px-2 py-1 text-xs"
             disabled={
               loading ||
+              roleLoading ||
               !rawWorkspacesData ||
               Object.keys(rawWorkspacesData).length === 0
             }
           >
-            Edit All Configs
+            {roleLoading ? (
+              <div className="flex items-center">
+                <CircularProgress size={12} className="mr-1" />
+                <span>Edit All Configs</span>
+              </div>
+            ) : (
+              'Edit All Configs'
+            )}
           </Button>
         </div>
         <div className="flex items-center">
@@ -664,9 +729,16 @@ export function Workspaces() {
               onDelete={handleDeleteWorkspace}
               onEdit={(name) => router.push(`/workspaces/${name}`)}
               router={router}
+              rawWorkspacesData={rawWorkspacesData}
+              checkPermissionAndAct={checkPermissionAndAct}
+              roleLoading={roleLoading}
             />
           ))}
-          <CreateWorkspaceCard onClick={() => router.push('/workspace/new')} />
+          <CreateWorkspaceCard
+            onClick={() => router.push('/workspace/new')}
+            checkPermissionAndAct={checkPermissionAndAct}
+            roleLoading={roleLoading}
+          />
         </div>
       )}
 
@@ -691,6 +763,50 @@ export function Workspaces() {
         </Dialog>
       )}
 
+      {/* Permission Denial Dialog */}
+      <Dialog
+        open={permissionDenialState.open}
+        onOpenChange={(open) =>
+          setPermissionDenialState((prev) => ({ ...prev, open }))
+        }
+      >
+        <DialogContent className="sm:max-w-md transition-all duration-200 ease-in-out">
+          <DialogHeader>
+            <DialogTitle>Permission Denied</DialogTitle>
+            <DialogDescription>
+              {roleLoading ? (
+                <div className="flex items-center py-2">
+                  <CircularProgress size={16} className="mr-2" />
+                  <span>Checking permissions...</span>
+                </div>
+              ) : (
+                <>
+                  {permissionDenialState.userName ? (
+                    <>
+                      {permissionDenialState.userName} is logged in as non-admin
+                      and {permissionDenialState.message}.
+                    </>
+                  ) : (
+                    permissionDenialState.message
+                  )}
+                </>
+              )}
+            </DialogDescription>
+          </DialogHeader>
+          <DialogFooter>
+            <Button
+              variant="outline"
+              onClick={() =>
+                setPermissionDenialState((prev) => ({ ...prev, open: false }))
+              }
+              disabled={roleLoading}
+            >
+              OK
+            </Button>
+          </DialogFooter>
+        </DialogContent>
+      </Dialog>
+
       {/* Delete Confirmation Dialog */}
       <Dialog open={deleteState.confirmOpen} onOpenChange={handleCancelDelete}>
         <DialogContent className="sm:max-w-md">
diff --git a/sky/dashboard/src/data/connectors/jobs.jsx b/sky/dashboard/src/data/connectors/jobs.jsx
index de62613cd..0e914b5bb 100644
--- a/sky/dashboard/src/data/connectors/jobs.jsx
+++ b/sky/dashboard/src/data/connectors/jobs.jsx
@@ -153,7 +153,7 @@ export async function getManagedJobs({ allUsers = true } = {}) {
           ? new Date(job.submitted_at * 1000)
           : null,
         events: events,
-        dag_yaml: job.dag_yaml,
+        dag_yaml: job.user_yaml,
         entrypoint: job.entrypoint,
       };
     });
@@ -311,7 +311,18 @@ export async function streamManagedJobLogs({
           onNewLog(chunk);
         }
       } finally {
-        reader.cancel();
+        // Only cancel the reader if the signal hasn't been aborted
+        // If signal is aborted, the reader should already be canceling
+        if (!signal || !signal.aborted) {
+          try {
+            reader.cancel();
+          } catch (cancelError) {
+            // Ignore errors from reader cancellation
+            if (cancelError.name !== 'AbortError') {
+              console.warn('Error canceling reader:', cancelError);
+            }
+          }
+        }
         // Clear the timeout when streaming completes successfully
         if (timeoutId) {
           clearTimeout(timeoutId);
diff --git a/sky/dashboard/src/data/connectors/users.js b/sky/dashboard/src/data/connectors/users.js
index ea1be4bcf..b149311a2 100644
--- a/sky/dashboard/src/data/connectors/users.js
+++ b/sky/dashboard/src/data/connectors/users.js
@@ -3,11 +3,19 @@ import { getClusters } from './clusters';
 import { getManagedJobs } from './jobs';
 
 // Helper functions for username parsing
-const parseUsername = (username) => {
+const parseUsername = (username, userId) => {
   if (username && username.includes('@')) {
     return username.split('@')[0];
   }
-  return username || 'N/A';
+  // If no email, show username with userId in parentheses only if they're different
+  const usernameBase = username || 'N/A';
+
+  // Skip showing userId if it's the same as username
+  if (userId && userId !== usernameBase) {
+    return `${usernameBase} (${userId})`;
+  }
+
+  return usernameBase;
 };
 
 const getFullEmail = (username) => {
@@ -68,7 +76,7 @@ export async function getUsersWithCounts() {
     );
     return {
       ...user,
-      usernameDisplay: parseUsername(user.username),
+      usernameDisplay: parseUsername(user.username, user.userId),
       fullEmail: getFullEmail(user.username),
       clusterCount: userClusters.length,
       jobCount: userJobs.length,
diff --git a/sky/dashboard/src/pages/config.js b/sky/dashboard/src/pages/config.js
index b45298d1b..4d8615abe 100644
--- a/sky/dashboard/src/pages/config.js
+++ b/sky/dashboard/src/pages/config.js
@@ -1,4 +1,4 @@
-import React, { useState, useEffect } from 'react';
+import React, { useState, useEffect, useRef } from 'react';
 import { useRouter } from 'next/router';
 import { Layout } from '@/components/elements/layout';
 import Link from 'next/link';
@@ -20,11 +20,21 @@ export default function ConfigPage() {
   const [saving, setSaving] = useState(false);
   const [error, setError] = useState(null);
   const [saveSuccess, setSaveSuccess] = useState(false);
+  const successTimeoutRef = useRef(null);
 
   useEffect(() => {
     loadConfig();
   }, []);
 
+  // Cleanup timeout on unmount
+  useEffect(() => {
+    return () => {
+      if (successTimeoutRef.current) {
+        clearTimeout(successTimeoutRef.current);
+      }
+    };
+  }, []);
+
   const loadConfig = async () => {
     setLoading(true);
     setError(null);
@@ -46,7 +56,12 @@ export default function ConfigPage() {
   const handleSave = async () => {
     setSaving(true);
     setError(null);
-    setSaveSuccess(false);
+    // Clear any existing success timeout
+    if (successTimeoutRef.current) {
+      clearTimeout(successTimeoutRef.current);
+      successTimeoutRef.current = null;
+    }
+
     try {
       // Check user role first
       const response = await fetch(`${ENDPOINT}/users/role`);
@@ -59,7 +74,9 @@ export default function ConfigPage() {
 
       if (currentUserRole != 'admin') {
         setError(
-          new Error(`${data.name} is logged in as no admin, cannot edit config`)
+          new Error(
+            `${data.name} is logged in as non-admin and cannot edit config`
+          )
         );
         setSaving(false);
         return;
@@ -92,8 +109,11 @@ export default function ConfigPage() {
 
       await updateConfig(parsedConfig);
       setSaveSuccess(true);
-      // Auto-hide success message after 3 seconds
-      setTimeout(() => setSaveSuccess(false), 3000);
+      // Auto-hide success message after 5 seconds
+      successTimeoutRef.current = setTimeout(() => {
+        setSaveSuccess(false);
+        successTimeoutRef.current = null;
+      }, 5000);
     } catch (error) {
       console.error('Error saving config:', error);
       setError(error);
@@ -110,6 +130,14 @@ export default function ConfigPage() {
     loadConfig();
   };
 
+  const handleSuccessDismiss = () => {
+    setSaveSuccess(false);
+    if (successTimeoutRef.current) {
+      clearTimeout(successTimeoutRef.current);
+      successTimeoutRef.current = null;
+    }
+  };
+
   return (
     <>
       <Head>
@@ -160,24 +188,48 @@ export default function ConfigPage() {
             {/* Success Message */}
             {saveSuccess && (
               <div className="bg-green-50 border border-green-200 rounded p-4 mb-6">
-                <div className="flex items-center">
-                  <div className="flex-shrink-0">
-                    <svg
-                      className="h-5 w-5 text-green-400"
-                      viewBox="0 0 20 20"
-                      fill="currentColor"
-                    >
-                      <path
-                        fillRule="evenodd"
-                        d="M10 18a8 8 0 100-16 8 8 0 000 16zm3.707-9.293a1 1 0 00-1.414-1.414L9 10.586 7.707 9.293a1 1 0 00-1.414 1.414l2 2a1 1 0 001.414 0l4-4z"
-                        clipRule="evenodd"
-                      />
-                    </svg>
+                <div className="flex items-center justify-between">
+                  <div className="flex items-center">
+                    <div className="flex-shrink-0">
+                      <svg
+                        className="h-5 w-5 text-green-400"
+                        viewBox="0 0 20 20"
+                        fill="currentColor"
+                      >
+                        <path
+                          fillRule="evenodd"
+                          d="M10 18a8 8 0 100-16 8 8 0 000 16zm3.707-9.293a1 1 0 00-1.414-1.414L9 10.586 7.707 9.293a1 1 0 00-1.414 1.414l2 2a1 1 0 001.414 0l4-4z"
+                          clipRule="evenodd"
+                        />
+                      </svg>
+                    </div>
+                    <div className="ml-3">
+                      <p className="text-sm font-medium text-green-800">
+                        Configuration saved successfully!
+                      </p>
+                    </div>
                   </div>
-                  <div className="ml-3">
-                    <p className="text-sm font-medium text-green-800">
-                      Configuration saved successfully!
-                    </p>
+                  <div className="ml-auto pl-3">
+                    <div className="-mx-1.5 -my-1.5">
+                      <button
+                        type="button"
+                        onClick={handleSuccessDismiss}
+                        className="inline-flex rounded-md bg-green-50 p-1.5 text-green-500 hover:bg-green-100 focus:outline-none focus:ring-2 focus:ring-green-600 focus:ring-offset-2 focus:ring-offset-green-50"
+                      >
+                        <span className="sr-only">Dismiss</span>
+                        <svg
+                          className="h-5 w-5"
+                          viewBox="0 0 20 20"
+                          fill="currentColor"
+                        >
+                          <path
+                            fillRule="evenodd"
+                            d="M4.293 4.293a1 1 0 011.414 0L10 8.586l4.293-4.293a1 1 0 111.414 1.414L11.414 10l4.293 4.293a1 1 0 01-1.414 1.414L10 11.414l-4.293 4.293a1 1 0 01-1.414-1.414L8.586 10 4.293 5.707a1 1 0 010-1.414z"
+                            clipRule="evenodd"
+                          />
+                        </svg>
+                      </button>
+                    </div>
                   </div>
                 </div>
               </div>
diff --git a/sky/dashboard/src/pages/jobs/[job].js b/sky/dashboard/src/pages/jobs/[job].js
index 480157ad6..564d7f922 100755
--- a/sky/dashboard/src/pages/jobs/[job].js
+++ b/sky/dashboard/src/pages/jobs/[job].js
@@ -339,6 +339,9 @@ function JobDetailsContent({
   const activeLogsRequest = useRef(null);
   const activeControllerLogsRequest = useRef(null);
 
+  // Track aborted controllers to prevent double-abort
+  const abortedControllers = useRef(new WeakSet());
+
   // Auto-scroll refs
   const logsContainerRef = useRef(null);
   const controllerLogsContainerRef = useRef(null);
@@ -353,6 +356,56 @@ function JobDetailsContent({
   const BATCH_UPDATE_INTERVAL = 100; // Batch updates every 100ms
   const THROTTLE_INTERVAL = 50; // Minimum time between updates
 
+  // Helper function to safely abort an AbortController
+  const safeAbort = useCallback((controller, description = 'request') => {
+    if (!controller) return;
+
+    // Check if we've already aborted this controller
+    if (abortedControllers.current.has(controller)) {
+      console.log(
+        `${description} controller already aborted previously, skipping`
+      );
+      return;
+    }
+
+    try {
+      // Additional safety checks
+      if (typeof controller.abort !== 'function') {
+        console.warn(
+          `Controller for ${description} does not have abort method`
+        );
+        return;
+      }
+
+      // Check if already aborted via signal
+      if (controller.signal && controller.signal.aborted) {
+        console.log(`${description} already aborted via signal, skipping`);
+        abortedControllers.current.add(controller); // Mark as aborted
+        return;
+      }
+
+      // Attempt to abort
+      controller.abort();
+      abortedControllers.current.add(controller); // Mark as aborted
+      console.log(`Successfully aborted ${description}`);
+    } catch (error) {
+      // Handle any type of error that might occur during abort
+      console.log(
+        `Caught error while aborting ${description}:`,
+        error.name,
+        error.message
+      );
+
+      // Mark as aborted even if there was an error to prevent retry
+      abortedControllers.current.add(controller);
+
+      // Only warn for unexpected errors (not AbortError or InvalidStateError)
+      if (error.name !== 'AbortError' && error.name !== 'InvalidStateError') {
+        console.warn(`Unexpected error aborting ${description}:`, error);
+      }
+    }
+  }, []);
+
   // Custom hook for auto-scrolling
   const scrollToBottom = useCallback((logType) => {
     const containerRef =
@@ -665,9 +718,9 @@ function JobDetailsContent({
             console.log(`Cleaning up ${logType} request`);
             active = false;
 
-            // Abort the controller if it matches the current one
+            // Safely abort the controller if it matches the current one
             if (activeRequestRef.current === controller) {
-              activeRequestRef.current.abort();
+              safeAbort(controller, logType);
               activeRequestRef.current = null;
             }
 
@@ -690,9 +743,9 @@ function JobDetailsContent({
           console.log(`Cleaning up ${logType} request`);
           active = false;
 
-          // Abort the controller if it matches the current one
+          // Safely abort the controller if it matches the current one and hasn't been cleaned up yet
           if (activeRequestRef.current === controller) {
-            activeRequestRef.current.abort();
+            safeAbort(controller, `${logType} cleanup`);
             activeRequestRef.current = null;
           }
 
@@ -714,14 +767,14 @@ function JobDetailsContent({
         active = false;
       };
     },
-    [isPending, isPreStart, isRecovering, hasReceivedFirstChunk]
+    [isPending, isPreStart, isRecovering, hasReceivedFirstChunk, safeAbort]
   );
 
   // Fetch both logs and controller logs in parallel, regardless of activeTab
   useEffect(() => {
     // Cancel any existing request
     if (activeLogsRequest.current) {
-      activeLogsRequest.current.abort();
+      safeAbort(activeLogsRequest.current, 'logs');
       activeLogsRequest.current = null;
     }
 
@@ -730,12 +783,19 @@ function JobDetailsContent({
       const cleanup = fetchLogs('logs', jobData.id, setLogs, setIsLoadingLogs);
       return cleanup;
     }
-  }, [jobData.id, fetchLogs, setIsLoadingLogs, isPending, isRecovering]);
+  }, [
+    jobData.id,
+    fetchLogs,
+    setIsLoadingLogs,
+    isPending,
+    isRecovering,
+    safeAbort,
+  ]);
 
   useEffect(() => {
     // Cancel any existing request
     if (activeControllerLogsRequest.current) {
-      activeControllerLogsRequest.current.abort();
+      safeAbort(activeControllerLogsRequest.current, 'controller logs');
       activeControllerLogsRequest.current = null;
     }
 
@@ -749,14 +809,20 @@ function JobDetailsContent({
       );
       return cleanup;
     }
-  }, [jobData.id, fetchLogs, setIsLoadingControllerLogs, isPreStart]);
+  }, [
+    jobData.id,
+    fetchLogs,
+    setIsLoadingControllerLogs,
+    isPreStart,
+    safeAbort,
+  ]);
 
   // Handle refresh for logs
   useEffect(() => {
     if (refreshFlag > 0 && activeTab === 'logs') {
       // Cancel any existing request
       if (activeLogsRequest.current) {
-        activeLogsRequest.current.abort();
+        safeAbort(activeLogsRequest.current, 'logs refresh');
         activeLogsRequest.current = null;
       }
 
@@ -770,14 +836,24 @@ function JobDetailsContent({
       );
       return cleanup;
     }
-  }, [refreshFlag, activeTab, jobData.id, fetchLogs, setIsLoadingLogs]);
+  }, [
+    refreshFlag,
+    activeTab,
+    jobData.id,
+    fetchLogs,
+    setIsLoadingLogs,
+    safeAbort,
+  ]);
 
   // Handle refresh for controller logs
   useEffect(() => {
     if (refreshFlag > 0 && activeTab === 'controllerlogs') {
       // Cancel any existing request
       if (activeControllerLogsRequest.current) {
-        activeControllerLogsRequest.current.abort();
+        safeAbort(
+          activeControllerLogsRequest.current,
+          'controller logs refresh'
+        );
         activeControllerLogsRequest.current = null;
       }
 
@@ -797,6 +873,7 @@ function JobDetailsContent({
     jobData.id,
     fetchLogs,
     setIsLoadingControllerLogs,
+    safeAbort,
   ]);
 
   // Comprehensive cleanup when component unmounts or user navigates away
@@ -806,12 +883,15 @@ function JobDetailsContent({
 
       // Abort all active log requests
       if (activeLogsRequest.current) {
-        activeLogsRequest.current.abort();
+        safeAbort(activeLogsRequest.current, 'logs cleanup');
         activeLogsRequest.current = null;
       }
 
       if (activeControllerLogsRequest.current) {
-        activeControllerLogsRequest.current.abort();
+        safeAbort(
+          activeControllerLogsRequest.current,
+          'controller logs cleanup'
+        );
         activeControllerLogsRequest.current = null;
       }
 
@@ -831,7 +911,7 @@ function JobDetailsContent({
       setIsRefreshingLogs(false);
       setIsRefreshingControllerLogs(false);
     };
-  }, []); // Empty dependency array means this runs only on unmount
+  }, [safeAbort]); // Empty dependency array means this runs only on unmount
 
   // Handle page visibility changes to pause streaming when tab is not active
   useEffect(() => {
@@ -843,11 +923,11 @@ function JobDetailsContent({
         // Uncomment if you want to completely stop streaming when tab is not visible
         /*
         if (activeLogsRequest.current) {
-          activeLogsRequest.current.abort();
+          safeAbort(activeLogsRequest.current, 'logs visibility');
           activeLogsRequest.current = null;
         }
         if (activeControllerLogsRequest.current) {
-          activeControllerLogsRequest.current.abort();
+          safeAbort(activeControllerLogsRequest.current, 'controller logs visibility');
           activeControllerLogsRequest.current = null;
         }
         */
@@ -972,7 +1052,7 @@ function JobDetailsContent({
         <div className="text-base mt-1 flex items-center">
           <StatusBadge status={jobData.status} />
           {jobData.priority && (
-            <span className="ml-2"> (priority: {jobData.priority})</span>
+            <span className="ml-2"> (Priority: {jobData.priority})</span>
           )}
         </div>
       </div>
diff --git a/sky/exceptions.py b/sky/exceptions.py
index 1fc634805..a9eb805fa 100644
--- a/sky/exceptions.py
+++ b/sky/exceptions.py
@@ -580,6 +580,29 @@ class JobExitCode(enum.IntEnum):
         return cls.FAILED
 
 
+class ExecutionRetryableError(Exception):
+    """Raised when task execution fails and should be retried."""
+
+    def __init__(self, message: str, hint: str,
+                 retry_wait_seconds: int) -> None:
+        super().__init__(message)
+        self.hint = hint
+        self.retry_wait_seconds = retry_wait_seconds
+
+    def __reduce__(self):
+        # Make sure the exception is picklable
+        return (self.__class__, (str(self), self.hint, self.retry_wait_seconds))
+
+
+class ExecutionPoolFullError(Exception):
+    """Raised when the execution pool is full."""
+
+
 class RequestAlreadyExistsError(Exception):
     """Raised when a request is already exists."""
     pass
+
+
+class PermissionDeniedError(Exception):
+    """Raised when a user does not have permission to access a resource."""
+    pass
diff --git a/sky/global_user_state.py b/sky/global_user_state.py
index 04c2d1046..51ac885d8 100644
--- a/sky/global_user_state.py
+++ b/sky/global_user_state.py
@@ -258,7 +258,8 @@ def create_table():
             'user_hash',
             sqlalchemy.Text(),
             default_statement='DEFAULT NULL',
-            value_to_replace_existing_entries=common_utils.get_user_hash())
+            value_to_replace_existing_entries=common_utils.get_current_user(
+            ).id)
         db_utils.add_column_to_table_sqlalchemy(
             session,
             'clusters',
@@ -370,11 +371,11 @@ def add_or_update_user(user: models.User) -> bool:
             raise ValueError('Unsupported database dialect')
 
 
-def get_user(user_id: str) -> models.User:
+def get_user(user_id: str) -> Optional[models.User]:
     with orm.Session(SQLALCHEMY_ENGINE) as session:
         row = session.query(user_table).filter_by(id=user_id).first()
     if row is None:
-        return models.User(id=user_id)
+        return None
     return models.User(id=row.id, name=row.name)
 
 
@@ -436,7 +437,7 @@ def add_or_update_cluster(cluster_name: str,
             cluster_launched_at = int(time.time())
         usage_intervals.append((cluster_launched_at, None))
 
-    user_hash = common_utils.get_user_hash()
+    user_hash = common_utils.get_current_user().id
     active_workspace = skypilot_config.get_active_workspace()
 
     conditional_values = {}
@@ -848,6 +849,8 @@ def get_cluster_from_name(
     if row is None:
         return None
     user_hash = _get_user_hash_or_current_user(row.user_hash)
+    user = get_user(user_hash)
+    user_name = user.name if user is not None else None
     # TODO: use namedtuple instead of dict
     record = {
         'name': row.name,
@@ -865,7 +868,7 @@ def get_cluster_from_name(
         'cluster_ever_up': bool(row.cluster_ever_up),
         'status_updated_at': row.status_updated_at,
         'user_hash': user_hash,
-        'user_name': get_user(user_hash).name,
+        'user_name': user_name,
         'config_hash': row.config_hash,
         'workspace': row.workspace,
         'last_creation_yaml': row.last_creation_yaml,
@@ -882,6 +885,8 @@ def get_clusters() -> List[Dict[str, Any]]:
     records = []
     for row in rows:
         user_hash = _get_user_hash_or_current_user(row.user_hash)
+        user = get_user(user_hash)
+        user_name = user.name if user is not None else None
         # TODO: use namedtuple instead of dict
         record = {
             'name': row.name,
@@ -899,7 +904,7 @@ def get_clusters() -> List[Dict[str, Any]]:
             'cluster_ever_up': bool(row.cluster_ever_up),
             'status_updated_at': row.status_updated_at,
             'user_hash': user_hash,
-            'user_name': get_user(user_hash).name,
+            'user_name': user_name,
             'config_hash': row.config_hash,
             'workspace': row.workspace,
             'last_creation_yaml': row.last_creation_yaml,
diff --git a/sky/jobs/scheduler.py b/sky/jobs/scheduler.py
index 617c5b790..ee88511cd 100644
--- a/sky/jobs/scheduler.py
+++ b/sky/jobs/scheduler.py
@@ -191,8 +191,8 @@ def maybe_schedule_next_jobs() -> None:
         pass
 
 
-def submit_job(job_id: int, dag_yaml_path: str, env_file_path: str,
-               priority: int) -> None:
+def submit_job(job_id: int, dag_yaml_path: str, original_user_yaml_path: str,
+               env_file_path: str, priority: int) -> None:
     """Submit an existing job to the scheduler.
 
     This should be called after a job is created in the `spot` table as
@@ -203,7 +203,8 @@ def submit_job(job_id: int, dag_yaml_path: str, env_file_path: str,
     The user hash should be set (e.g. via SKYPILOT_USER_ID) before calling this.
     """
     with filelock.FileLock(_get_lock_path()):
-        state.scheduler_set_waiting(job_id, dag_yaml_path, env_file_path,
+        state.scheduler_set_waiting(job_id, dag_yaml_path,
+                                    original_user_yaml_path, env_file_path,
                                     common_utils.get_user_hash(), priority)
     maybe_schedule_next_jobs()
 
@@ -312,6 +313,9 @@ if __name__ == '__main__':
     parser.add_argument('dag_yaml',
                         type=str,
                         help='The path to the user job yaml file.')
+    parser.add_argument('--user-yaml-path',
+                        type=str,
+                        help='The path to the original user job yaml file.')
     parser.add_argument('--job-id',
                         required=True,
                         type=int,
@@ -325,4 +329,5 @@ if __name__ == '__main__':
         default=500,
         help='Job priority (0-1000, lower is higher). Default: 500.')
     args = parser.parse_args()
-    submit_job(args.job_id, args.dag_yaml, args.env_file, args.priority)
+    submit_job(args.job_id, args.dag_yaml, args.user_yaml_path, args.env_file,
+               args.priority)
diff --git a/sky/jobs/server/core.py b/sky/jobs/server/core.py
index 9020d02bb..e6245eeac 100644
--- a/sky/jobs/server/core.py
+++ b/sky/jobs/server/core.py
@@ -37,6 +37,7 @@ from sky.utils import status_lib
 from sky.utils import subprocess_utils
 from sky.utils import timeline
 from sky.utils import ux_utils
+from sky.workspaces import core as workspaces_core
 
 if typing.TYPE_CHECKING:
     import sky
@@ -88,6 +89,9 @@ def launch(
             raise ValueError('Only single-task or chain DAG is '
                              f'allowed for job_launch. Dag: {dag}')
     dag.validate()
+
+    user_dag_str = dag_utils.dump_chain_dag_to_yaml_str(dag)
+
     dag_utils.maybe_infer_and_fill_dag_and_task_names(dag)
 
     task_names = set()
@@ -175,12 +179,20 @@ def launch(
                     controller_utils.translate_local_file_mounts_to_two_hop(
                         task_))
 
+    # Has to use `\` to avoid yapf issue.
     with tempfile.NamedTemporaryFile(prefix=f'managed-dag-{dag.name}-',
-                                     mode='w') as f:
+                                     mode='w') as f, \
+         tempfile.NamedTemporaryFile(prefix=f'managed-user-dag-{dag.name}-',
+                                     mode='w') as original_user_yaml_path:
+        original_user_yaml_path.write(user_dag_str)
+        original_user_yaml_path.flush()
+
         dag_utils.dump_chain_dag_to_yaml(dag, f.name)
         controller = controller_utils.Controllers.JOBS_CONTROLLER
         controller_name = controller.value.cluster_name
         prefix = managed_job_constants.JOBS_TASK_YAML_PREFIX
+        remote_original_user_yaml_path = (
+            f'{prefix}/{dag.name}-{dag_uuid}.original_user_yaml')
         remote_user_yaml_path = f'{prefix}/{dag.name}-{dag_uuid}.yaml'
         remote_user_config_path = f'{prefix}/{dag.name}-{dag_uuid}.config_yaml'
         remote_env_file_path = f'{prefix}/{dag.name}-{dag_uuid}.env'
@@ -189,6 +201,8 @@ def launch(
             task_resources=sum([list(t.resources) for t in dag.tasks], []))
 
         vars_to_fill = {
+            'remote_original_user_yaml_path': remote_original_user_yaml_path,
+            'original_user_dag_path': original_user_yaml_path.name,
             'remote_user_yaml_path': remote_user_yaml_path,
             'user_yaml_path': f.name,
             'local_to_controller_file_mounts': local_to_controller_file_mounts,
@@ -231,7 +245,7 @@ def launch(
 
         # Launch with the api server's user hash, so that sky status does not
         # show the owner of the controller as whatever user launched it first.
-        with common.with_server_user_hash():
+        with common.with_server_user():
             # Always launch the controller in the default workspace.
             with skypilot_config.local_active_workspace_ctx(
                     skylet_constants.SKYPILOT_DEFAULT_WORKSPACE):
@@ -442,6 +456,13 @@ def queue(refresh: bool,
 
         jobs = list(filter(user_hash_matches_or_missing, jobs))
 
+    accessible_workspaces = workspaces_core.get_workspaces()
+    jobs = list(
+        filter(
+            lambda job: job.get('workspace', skylet_constants.
+                                SKYPILOT_DEFAULT_WORKSPACE) in
+            accessible_workspaces, jobs))
+
     if skip_finished:
         # Filter out the finished jobs. If a multi-task job is partially
         # finished, we will include all its tasks.
diff --git a/sky/jobs/server/server.py b/sky/jobs/server/server.py
index 0e18ac072..919f91865 100644
--- a/sky/jobs/server/server.py
+++ b/sky/jobs/server/server.py
@@ -1,12 +1,9 @@
 """REST API for managed jobs."""
-import os
 
 import fastapi
-import httpx
 
 from sky import sky_logging
 from sky.jobs.server import core
-from sky.jobs.server import dashboard_utils
 from sky.server import common as server_common
 from sky.server import stream_utils
 from sky.server.requests import executor
@@ -14,7 +11,6 @@ from sky.server.requests import payloads
 from sky.server.requests import requests as api_requests
 from sky.skylet import constants
 from sky.utils import common
-from sky.utils import common_utils
 
 logger = sky_logging.init_logger(__name__)
 
@@ -110,94 +106,3 @@ async def download_logs(
         if jobs_download_logs_body.refresh else api_requests.ScheduleType.SHORT,
         request_cluster_name=common.JOB_CONTROLLER_NAME,
     )
-
-
-@router.get('/dashboard')
-async def dashboard(request: fastapi.Request,
-                    user_hash: str) -> fastapi.Response:
-    # TODO(cooperc): Support showing only jobs for a specific user.
-
-    # FIX(zhwu/cooperc/eric): Fix log downloading (assumes global
-    # /download_log/xx route)
-
-    # Note: before #4717, each user had their own controller, and thus their own
-    # dashboard. Now, all users share the same controller, so this isn't really
-    # necessary. TODO(cooperc): clean up.
-
-    # TODO: Put this in an executor to avoid blocking the main server thread.
-    # It can take a long time if it needs to check the controller status.
-
-    # Find the port for the dashboard of the user
-    os.environ[constants.USER_ID_ENV_VAR] = user_hash
-    server_common.reload_for_new_request(client_entrypoint=None,
-                                         client_command=None,
-                                         using_remote_api_server=False)
-    logger.info(f'Starting dashboard for user hash: {user_hash}')
-
-    with dashboard_utils.get_dashboard_lock_for_user(user_hash):
-        max_retries = 3
-        for attempt in range(max_retries):
-            port, pid = dashboard_utils.get_dashboard_session(user_hash)
-            if port == 0 or attempt > 0:
-                # Let the client know that we are waiting for starting the
-                # dashboard.
-                try:
-                    port, pid = core.start_dashboard_forwarding()
-                except Exception as e:  # pylint: disable=broad-except
-                    # We catch all exceptions to gracefully handle unknown
-                    # errors and raise an HTTPException to the client.
-                    msg = (
-                        'Dashboard failed to start: '
-                        f'{common_utils.format_exception(e, use_bracket=True)}')
-                    logger.error(msg)
-                    raise fastapi.HTTPException(status_code=503, detail=msg)
-                dashboard_utils.add_dashboard_session(user_hash, port, pid)
-
-            # Assuming the dashboard is forwarded to localhost on the API server
-            dashboard_url = f'http://localhost:{port}'
-            try:
-                # Ping the dashboard to check if it's still running
-                async with httpx.AsyncClient() as client:
-                    response = await client.request('GET',
-                                                    dashboard_url,
-                                                    timeout=5)
-                if response.is_success:
-                    break  # Connection successful, proceed with the request
-                # Raise an HTTPException here which will be caught by the
-                # following except block to retry with new connection
-                response.raise_for_status()
-            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)}')
-                logger.info(msg)
-                if attempt == max_retries - 1:
-                    raise fastapi.HTTPException(status_code=503, detail=msg)
-
-    # Create a client session to forward the request
-    try:
-        async with httpx.AsyncClient() as client:
-            # Make the request and get the response
-            response = await client.request(
-                method='GET',
-                url=f'{dashboard_url}',
-                headers=request.headers.raw,
-            )
-
-            # Create a new response with the content already read
-            content = await response.aread()
-            return fastapi.Response(
-                content=content,
-                status_code=response.status_code,
-                headers=dict(response.headers),
-                media_type=response.headers.get('content-type'))
-    except Exception as e:
-        msg = (f'Failed to forward request to dashboard: '
-               f'{common_utils.format_exception(e, use_bracket=True)}')
-        logger.error(msg)
-        raise fastapi.HTTPException(status_code=502, detail=msg)
diff --git a/sky/jobs/state.py b/sky/jobs/state.py
index 2a030d93b..a863aeeae 100644
--- a/sky/jobs/state.py
+++ b/sky/jobs/state.py
@@ -122,7 +122,8 @@ def create_table(cursor, conn):
         user_hash TEXT,
         workspace TEXT DEFAULT NULL,
         priority INTEGER DEFAULT 500,
-        entrypoint TEXT DEFAULT NULL)""")
+        entrypoint TEXT DEFAULT NULL,
+        original_user_yaml_path TEXT DEFAULT NULL)""")
 
     db_utils.add_column_to_table(cursor, conn, 'job_info', 'schedule_state',
                                  'TEXT')
@@ -153,6 +154,8 @@ def create_table(cursor, conn):
                                  value_to_replace_existing_entries=500)
 
     db_utils.add_column_to_table(cursor, conn, 'job_info', 'entrypoint', 'TEXT')
+    db_utils.add_column_to_table(cursor, conn, 'job_info',
+                                 'original_user_yaml_path', 'TEXT')
     conn.commit()
 
 
@@ -212,6 +215,7 @@ columns = [
     'workspace',
     'priority',
     'entrypoint',
+    'original_user_yaml_path',
 ]
 
 
@@ -1013,19 +1017,16 @@ def get_managed_jobs(job_id: Optional[int] = None) -> List[Dict[str, Any]]:
             if job_dict['job_name'] is None:
                 job_dict['job_name'] = job_dict['task_name']
 
-            # Add YAML content and command for managed jobs
-            dag_yaml_path = job_dict.get('dag_yaml_path')
-            if dag_yaml_path:
+            # Add user YAML content for managed jobs.
+            yaml_path = job_dict.get('original_user_yaml_path')
+            if yaml_path:
                 try:
-                    with open(dag_yaml_path, 'r', encoding='utf-8') as f:
-                        job_dict['dag_yaml'] = f.read()
+                    with open(yaml_path, 'r', encoding='utf-8') as f:
+                        job_dict['user_yaml'] = f.read()
                 except (FileNotFoundError, IOError, OSError):
-                    job_dict['dag_yaml'] = None
-
-                # Generate a command that could be used to launch this job
-                # Format: sky jobs launch <yaml_path>
+                    job_dict['user_yaml'] = None
             else:
-                job_dict['dag_yaml'] = None
+                job_dict['user_yaml'] = None
 
             jobs.append(job_dict)
         return jobs
@@ -1083,18 +1084,20 @@ def get_local_log_file(job_id: int, task_id: Optional[int]) -> Optional[str]:
 # scheduler lock to work correctly.
 
 
-def scheduler_set_waiting(job_id: int, dag_yaml_path: str, env_file_path: str,
+def scheduler_set_waiting(job_id: int, dag_yaml_path: str,
+                          original_user_yaml_path: str, env_file_path: str,
                           user_hash: str, priority: int) -> None:
     """Do not call without holding the scheduler lock."""
     with db_utils.safe_cursor(_DB_PATH) as cursor:
         updated_count = cursor.execute(
             'UPDATE job_info SET '
-            'schedule_state = (?), dag_yaml_path = (?), env_file_path = (?), '
+            'schedule_state = (?), dag_yaml_path = (?), '
+            'original_user_yaml_path = (?), env_file_path = (?), '
             '  user_hash = (?), priority = (?) '
             'WHERE spot_job_id = (?) AND schedule_state = (?)',
             (ManagedJobScheduleState.WAITING.value, dag_yaml_path,
-             env_file_path, user_hash, priority, job_id,
-             ManagedJobScheduleState.INACTIVE.value)).rowcount
+             original_user_yaml_path, env_file_path, user_hash, priority,
+             job_id, ManagedJobScheduleState.INACTIVE.value)).rowcount
         assert updated_count == 1, (job_id, updated_count)
 
 
diff --git a/sky/jobs/utils.py b/sky/jobs/utils.py
index 15036a489..9a6a914a7 100644
--- a/sky/jobs/utils.py
+++ b/sky/jobs/utils.py
@@ -1025,7 +1025,8 @@ def load_managed_job_queue(payload: str) -> List[Dict[str, Any]]:
         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
+            user = global_user_state.get_user(job['user_hash'])
+            job['user_name'] = user.name if user is not None else None
     return jobs
 
 
diff --git a/sky/models.py b/sky/models.py
index 1c9bfa45e..2ec2473a1 100644
--- a/sky/models.py
+++ b/sky/models.py
@@ -2,8 +2,13 @@
 
 import collections
 import dataclasses
+import getpass
+import os
 from typing import Any, Dict, Optional
 
+from sky.skylet import constants
+from sky.utils import common_utils
+
 
 @dataclasses.dataclass
 class User:
@@ -16,6 +21,19 @@ class User:
     def to_dict(self) -> Dict[str, Any]:
         return {'id': self.id, 'name': self.name}
 
+    def to_env_vars(self) -> Dict[str, Any]:
+        return {
+            constants.USER_ID_ENV_VAR: self.id,
+            constants.USER_ENV_VAR: self.name,
+        }
+
+    @classmethod
+    def get_current_user(cls) -> 'User':
+        """Returns the current user."""
+        user_name = os.getenv(constants.USER_ENV_VAR, getpass.getuser())
+        user_hash = common_utils.get_user_hash()
+        return User(id=user_hash, name=user_name)
+
 
 RealtimeGpuAvailability = collections.namedtuple(
     'RealtimeGpuAvailability', ['gpu', 'counts', 'capacity', 'available'])
diff --git a/sky/serve/server/core.py b/sky/serve/server/core.py
index 83b279670..82957cda9 100644
--- a/sky/serve/server/core.py
+++ b/sky/serve/server/core.py
@@ -221,7 +221,7 @@ def up(
         # for the first time; otherwise it is a name conflict.
         # Since the controller may be shared among multiple users, launch the
         # controller with the API server's user hash.
-        with common.with_server_user_hash():
+        with common.with_server_user():
             with skypilot_config.local_active_workspace_ctx(
                     constants.SKYPILOT_DEFAULT_WORKSPACE):
                 controller_job_id, controller_handle = execution.launch(
diff --git a/sky/server/common.py b/sky/server/common.py
index 9926842b7..96c3e0694 100644
--- a/sky/server/common.py
+++ b/sky/server/common.py
@@ -39,6 +39,7 @@ if typing.TYPE_CHECKING:
     import requests
 
     from sky import dag as dag_lib
+    from sky import models
 else:
     pydantic = adaptors_common.LazyImport('pydantic')
     requests = adaptors_common.LazyImport('requests')
@@ -710,7 +711,7 @@ def request_body_to_params(body: 'pydantic.BaseModel') -> Dict[str, Any]:
 
 def reload_for_new_request(client_entrypoint: Optional[str],
                            client_command: Optional[str],
-                           using_remote_api_server: bool):
+                           using_remote_api_server: bool, user: 'models.User'):
     """Reload modules, global variables, and usage message for a new request."""
     # This should be called first to make sure the logger is up-to-date.
     sky_logging.reload_logger()
@@ -719,10 +720,11 @@ def reload_for_new_request(client_entrypoint: Optional[str],
     skypilot_config.safe_reload_config()
 
     # Reset the client entrypoint and command for the usage message.
-    common_utils.set_client_status(
+    common_utils.set_request_context(
         client_entrypoint=client_entrypoint,
         client_command=client_command,
         using_remote_api_server=using_remote_api_server,
+        user=user,
     )
 
     # Clear cache should be called before reload_logger and usage reset,
diff --git a/sky/server/constants.py b/sky/server/constants.py
index 5a85dcd56..f8996972d 100644
--- a/sky/server/constants.py
+++ b/sky/server/constants.py
@@ -11,8 +11,6 @@ API_VERSION = '9'
 
 # Prefix for API request names.
 REQUEST_NAME_PREFIX = 'sky.'
-# The user ID of the SkyPilot system.
-SKYPILOT_SYSTEM_USER_ID = 'skypilot-system'
 # The memory (GB) that SkyPilot tries to not use to prevent OOM.
 MIN_AVAIL_MEM_GB = 2
 # Default encoder/decoder handler name.
diff --git a/sky/server/requests/executor.py b/sky/server/requests/executor.py
index 0b13ccc93..76360985b 100644
--- a/sky/server/requests/executor.py
+++ b/sky/server/requests/executor.py
@@ -19,6 +19,7 @@ The number of the workers is determined by the system resources.
 See the [README.md](../README.md) for detailed architecture of the executor.
 """
 import asyncio
+import concurrent.futures
 import contextlib
 import multiprocessing
 import os
@@ -53,6 +54,7 @@ from sky.utils import context
 from sky.utils import context_utils
 from sky.utils import subprocess_utils
 from sky.utils import timeline
+from sky.workspaces import core as workspaces_core
 
 if typing.TYPE_CHECKING:
     import types
@@ -92,21 +94,21 @@ class RequestQueue:
         else:
             raise RuntimeError(f'Invalid queue backend: {backend}')
 
-    def put(self, request: Tuple[str, bool]) -> None:
+    def put(self, request: Tuple[str, bool, bool]) -> None:
         """Put and request to the queue.
 
         Args:
-            request: A tuple of request_id and ignore_return_value.
+            request: A tuple of request_id, ignore_return_value, and retryable.
         """
         self.queue.put(request)  # type: ignore
 
-    def get(self) -> Optional[Tuple[str, bool]]:
+    def get(self) -> Optional[Tuple[str, bool, bool]]:
         """Get a request from the queue.
 
         It is non-blocking if the queue is empty, and returns None.
 
         Returns:
-            A tuple of request_id and ignore_return_value.
+            A tuple of request_id, ignore_return_value, and retryable.
         """
         try:
             return self.queue.get(block=False)
@@ -158,7 +160,7 @@ class RequestWorker:
             if request_element is None:
                 time.sleep(0.1)
                 return
-            request_id, ignore_return_value = request_element
+            request_id, ignore_return_value, retryable = request_element
             request = api_requests.get_request(request_id)
             assert request is not None, f'Request with ID {request_id} is None'
             if request.status == api_requests.RequestStatus.CANCELLED:
@@ -170,8 +172,14 @@ class RequestWorker:
             # multiple requests can share the same process pid, which may cause
             # issues with SkyPilot core functions if they rely on the exit of
             # the process, such as subprocess_daemon.py.
-            executor.submit_until_success(_request_execution_wrapper,
-                                          request_id, ignore_return_value)
+            fut = executor.submit_until_success(_request_execution_wrapper,
+                                                request_id, ignore_return_value)
+            if retryable:
+                # If the task might fail and be retried, start a thread to
+                # monitor the future and process retry.
+                threading.Thread(target=self.handle_task_result,
+                                 args=(fut, request_element),
+                                 daemon=True).start()
 
             logger.info(f'[{self}] Submitted request: {request_id}')
         except (Exception, SystemExit) as e:  # pylint: disable=broad-except
@@ -181,6 +189,16 @@ class RequestWorker:
                 f'{request_id if "request_id" in locals() else ""} '
                 f'{common_utils.format_exception(e, use_bracket=True)}')
 
+    def handle_task_result(self, fut: concurrent.futures.Future,
+                           request_element: Tuple[str, bool, bool]) -> None:
+        try:
+            fut.result()
+        except exceptions.ExecutionRetryableError as e:
+            time.sleep(e.retry_wait_seconds)
+            # Reschedule the request.
+            queue = _get_queue(self.schedule_type)
+            queue.put(request_element)
+
     def run(self) -> None:
         # Handle the SIGTERM signal to abort the executor process gracefully.
         proc_group = f'{self.schedule_type.value}'
@@ -229,6 +247,9 @@ def override_request_env_and_config(
     original_env = os.environ.copy()
     os.environ.update(request_body.env_vars)
     # Note: may be overridden by AuthProxyMiddleware.
+    # TODO(zhwu): we need to make the entire request a context available to the
+    # entire request execution, so that we can access info like user through
+    # the execution.
     user = models.User(id=request_body.env_vars[constants.USER_ID_ENV_VAR],
                        name=request_body.env_vars[constants.USER_ENV_VAR])
     global_user_state.add_or_update_user(user)
@@ -237,13 +258,17 @@ def override_request_env_and_config(
     server_common.reload_for_new_request(
         client_entrypoint=request_body.entrypoint,
         client_command=request_body.entrypoint_command,
-        using_remote_api_server=request_body.using_remote_api_server)
+        using_remote_api_server=request_body.using_remote_api_server,
+        user=user)
     try:
         logger.debug(
             f'override path: {request_body.override_skypilot_config_path}')
         with skypilot_config.override_skypilot_config(
                 request_body.override_skypilot_config,
                 request_body.override_skypilot_config_path):
+            # Rejecting requests to workspaces that the user does not have
+            # permission to access.
+            workspaces_core.reject_request_for_unauthorized_workspace(user)
             yield
     finally:
         # We need to call the save_timeline() since atexit will not be
@@ -308,7 +333,9 @@ def _request_execution_wrapper(request_id: str,
         func = request_task.entrypoint
         request_body = request_task.request_body
 
-    with log_path.open('w', encoding='utf-8') as f:
+    # Append to the log file instead of overwriting it since there might be
+    # logs from previous retries.
+    with log_path.open('a', encoding='utf-8') as f:
         # Store copies of the original stdout and stderr file descriptors
         original_stdout, original_stderr = _redirect_output(f)
         # Redirect the stdout/stderr before overriding the environment and
@@ -332,6 +359,17 @@ def _request_execution_wrapper(request_id: str,
             subprocess_utils.kill_children_processes()
             _restore_output(original_stdout, original_stderr)
             return
+        except exceptions.ExecutionRetryableError as e:
+            logger.error(e)
+            logger.info(e.hint)
+            with api_requests.update_request(request_id) as request_task:
+                assert request_task is not None, request_id
+                # Retried request will undergo rescheduling and a new execution,
+                # clear the pid of the request.
+                request_task.pid = None
+            # Yield control to the scheduler for uniform handling of retries.
+            _restore_output(original_stdout, original_stderr)
+            raise
         except (Exception, SystemExit) as e:  # pylint: disable=broad-except
             api_requests.set_request_failed(request_id, e)
             _restore_output(original_stdout, original_stderr)
@@ -433,7 +471,7 @@ def prepare_request(
     """Prepare a request for execution."""
     user_id = request_body.env_vars[constants.USER_ID_ENV_VAR]
     if is_skypilot_system:
-        user_id = server_constants.SKYPILOT_SYSTEM_USER_ID
+        user_id = constants.SKYPILOT_SYSTEM_USER_ID
         global_user_state.add_or_update_user(
             models.User(id=user_id, name=user_id))
     request = api_requests.Request(request_id=request_id,
@@ -455,17 +493,17 @@ def prepare_request(
     return request
 
 
-def schedule_request(
-        request_id: str,
-        request_name: str,
-        request_body: payloads.RequestBody,
-        func: Callable[P, Any],
-        request_cluster_name: Optional[str] = None,
-        ignore_return_value: bool = False,
-        schedule_type: api_requests.ScheduleType = (
-            api_requests.ScheduleType.LONG),
-        is_skypilot_system: bool = False,
-        precondition: Optional[preconditions.Precondition] = None) -> None:
+def schedule_request(request_id: str,
+                     request_name: str,
+                     request_body: payloads.RequestBody,
+                     func: Callable[P, Any],
+                     request_cluster_name: Optional[str] = None,
+                     ignore_return_value: bool = False,
+                     schedule_type: api_requests.ScheduleType = (
+                         api_requests.ScheduleType.LONG),
+                     is_skypilot_system: bool = False,
+                     precondition: Optional[preconditions.Precondition] = None,
+                     retryable: bool = False) -> None:
     """Enqueue a request to the request queue.
 
     Args:
@@ -490,7 +528,7 @@ def schedule_request(
                     request_cluster_name, schedule_type, is_skypilot_system)
 
     def enqueue():
-        input_tuple = (request_id, ignore_return_value)
+        input_tuple = (request_id, ignore_return_value, retryable)
         logger.info(f'Queuing request: {request_id}')
         _get_queue(schedule_type).put(input_tuple)
 
diff --git a/sky/server/requests/process.py b/sky/server/requests/process.py
index a70c75f17..a8a17eb2e 100644
--- a/sky/server/requests/process.py
+++ b/sky/server/requests/process.py
@@ -6,6 +6,7 @@ import threading
 import time
 from typing import Callable, Dict, Optional, Tuple
 
+from sky import exceptions
 from sky.utils import atomic
 from sky.utils import subprocess_utils
 
@@ -67,14 +68,24 @@ class PoolExecutor(concurrent.futures.ProcessPoolExecutor):
 
 
 # Define the worker function outside of the class to avoid pickling self
-def _disposable_worker(fn, initializer: Optional[Callable], initargs: Tuple,
-                       args, kwargs):
+def _disposable_worker(fn, initializer, initargs, result_queue, args, kwargs):
+    """The worker function that is used to run the task.
+
+    Args:
+        fn: The function to run.
+        initializer: The initializer function to run before running the task.
+        initargs: The arguments to pass to the initializer function.
+        result_queue: The queue to put the result and exception into.
+        args: The arguments to pass to the function.
+        kwargs: The keyword arguments to pass to the function.
+    """
     try:
         if initializer is not None:
             initializer(*initargs)
-        fn(*args, **kwargs)
+        result = fn(*args, **kwargs)
+        result_queue.put(result)
     except BaseException as e:  # pylint: disable=broad-except
-        return e
+        result_queue.put(e)
 
 
 class DisposableExecutor:
@@ -98,28 +109,52 @@ class DisposableExecutor:
         self._initializer: Optional[Callable] = initializer
         self._initargs: Tuple = initargs
 
-    def _monitor_worker(self, process: multiprocessing.Process) -> None:
+    def _monitor_worker(self, process: multiprocessing.Process,
+                        future: concurrent.futures.Future,
+                        result_queue: multiprocessing.Queue) -> None:
         """Monitor the worker process and cleanup when it's done."""
-        process.join()
-        if process.pid:
-            with self._lock:
-                if process.pid in self.workers:
-                    del self.workers[process.pid]
-
-    # Submit is not compatible with ProcessPoolExecutor because we does not
-    # bother to return a Future. Can be improved if needed.
-    def submit(self, fn, *args, **kwargs) -> bool:
-        """Submit a task for execution."""
+        try:
+            process.join()
+            if not future.cancelled():
+                try:
+                    # Get result from the queue if process completed
+                    if not result_queue.empty():
+                        result = result_queue.get(block=False)
+                        if isinstance(result, BaseException):
+                            future.set_exception(result)
+                        else:
+                            future.set_result(result)
+                    else:
+                        # Process ended but no result
+                        future.set_result(None)
+                except (multiprocessing.TimeoutError, BrokenPipeError,
+                        EOFError) as e:
+                    future.set_exception(e)
+        finally:
+            if process.pid:
+                with self._lock:
+                    if process.pid in self.workers:
+                        del self.workers[process.pid]
+
+    def submit(self, fn, *args, **kwargs) -> concurrent.futures.Future:
+        """Submit a task for execution and return a Future."""
+        future: concurrent.futures.Future = concurrent.futures.Future()
+
         if self._shutdown:
-            return False
+            raise RuntimeError('Cannot submit task after executor is shutdown')
+
         with self._lock:
             if (self.max_workers is not None and
                     len(self.workers) >= self.max_workers):
-                return False
+                raise exceptions.ExecutionPoolFullError(
+                    'Maximum workers reached')
 
+        result_queue: multiprocessing.Queue = multiprocessing.Queue()
         process = multiprocessing.Process(target=_disposable_worker,
                                           args=(fn, self._initializer,
-                                                self._initargs, args, kwargs))
+                                                self._initargs, result_queue,
+                                                args, kwargs))
+        process.daemon = True
         process.start()
 
         with self._lock:
@@ -128,13 +163,13 @@ class DisposableExecutor:
                 raise RuntimeError('Failed to start process')
             self.workers[pid] = process
 
-        # Start monitor thread to cleanup the worker process when it's done.
+        # Start monitor thread to cleanup the worker process when it's done
         monitor_thread = threading.Thread(target=self._monitor_worker,
-                                          args=(process,),
+                                          args=(process, future, result_queue),
                                           daemon=True)
         monitor_thread.start()
 
-        return True
+        return future
 
     def has_idle_workers(self) -> bool:
         """Check if there are any idle workers."""
@@ -173,12 +208,14 @@ class BurstableExecutor:
             self._burst_executor = DisposableExecutor(max_workers=burst_workers,
                                                       **kwargs)
 
-    def submit_until_success(self, fn, *args, **kwargs):
+    def submit_until_success(self, fn, *args,
+                             **kwargs) -> concurrent.futures.Future:
         """Submit a task for execution until success.
 
         Prioritizes submitting to the guaranteed pool. If no idle workers
         are available in the guaranteed pool, it will submit to the burst
-        pool.
+        pool. If the burst pool is full, it will retry the whole process until
+        the task is submitted successfully.
         TODO(aylei): this is coupled with executor.RequestWorker since we
         know the worker is dedicated to request scheduling and it either
         blocks on request polling or request submitting. So it is no harm
@@ -188,17 +225,20 @@ class BurstableExecutor:
 
         while True:
             if self._executor is not None and self._executor.has_idle_workers():
-                self._executor.submit(fn, *args, **kwargs)
-                break
+                logger.info('Submitting to the guaranteed pool')
+                return self._executor.submit(fn, *args, **kwargs)
             if (self._burst_executor is not None and
                     self._burst_executor.has_idle_workers()):
-                self._burst_executor.submit(fn, *args, **kwargs)
-                break
+                try:
+                    fut = self._burst_executor.submit(fn, *args, **kwargs)
+                    return fut
+                except exceptions.ExecutionPoolFullError:
+                    # The burst pool is full, try the next candidate.
+                    pass
             if self._executor is not None:
                 # No idle workers in either pool, still queue the request
                 # to the guaranteed pool to keep behavior consistent.
-                self._executor.submit(fn, *args, **kwargs)
-                break
+                return self._executor.submit(fn, *args, **kwargs)
             logger.debug('No guaranteed pool set and the burst pool is full, '
                          'retry later.')
             time.sleep(0.1)
diff --git a/sky/server/requests/requests.py b/sky/server/requests/requests.py
index cce2c0138..173684811 100644
--- a/sky/server/requests/requests.py
+++ b/sky/server/requests/requests.py
@@ -11,7 +11,7 @@ import signal
 import sqlite3
 import time
 import traceback
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple
 
 import colorama
 import filelock
@@ -204,7 +204,8 @@ class Request:
         """
         assert isinstance(self.request_body,
                           payloads.RequestBody), (self.name, self.request_body)
-        user_name = global_user_state.get_user(self.user_id).name
+        user = global_user_state.get_user(self.user_id)
+        user_name = user.name if user is not None else None
         return RequestPayload(
             request_id=self.request_id,
             name=self.name,
@@ -464,7 +465,7 @@ def request_lock_path(request_id: str) -> str:
 
 @contextlib.contextmanager
 @init_db
-def update_request(request_id: str):
+def update_request(request_id: str) -> Generator[Optional[Request], None, None]:
     """Get a SkyPilot API request."""
     request = _get_request_no_lock(request_id)
     yield request
diff --git a/sky/server/server.py b/sky/server/server.py
index b83f08bb2..485673628 100644
--- a/sky/server/server.py
+++ b/sky/server/server.py
@@ -49,6 +49,7 @@ from sky.server.requests import preconditions
 from sky.server.requests import requests as requests_lib
 from sky.skylet import constants
 from sky.usage import usage_lib
+from sky.users import permission
 from sky.users import server as users_rest
 from sky.utils import admin_policy_utils
 from sky.utils import common as common_lib
@@ -105,17 +106,21 @@ class RBACMiddleware(starlette.middleware.base.BaseHTTPMiddleware):
     """Middleware to handle RBAC."""
 
     async def dispatch(self, request: fastapi.Request, call_next):
-        if request.url.path.startswith('/dashboard/'):
+        # TODO(hailong): should have a list of paths
+        # that are not checked for RBAC
+        if (request.url.path.startswith('/dashboard/') or
+                request.url.path.startswith('/api/')):
             return await call_next(request)
 
         auth_user = _get_auth_user_header(request)
         if auth_user is None:
             return await call_next(request)
 
-        permission_service = users_rest.permission_service
+        permission_service = permission.permission_service
         # Check the role permission
-        if permission_service.check_permission(auth_user.id, request.url.path,
-                                               request.method):
+        if permission_service.check_endpoint_permission(auth_user.id,
+                                                        request.url.path,
+                                                        request.method):
             return fastapi.responses.JSONResponse(
                 status_code=403, content={'detail': 'Forbidden'})
 
@@ -154,9 +159,15 @@ class AuthProxyMiddleware(starlette.middleware.base.BaseHTTPMiddleware):
         if auth_user is not None:
             newly_added = global_user_state.add_or_update_user(auth_user)
             if newly_added:
-                users_rest.permission_service.add_user_if_not_exists(
+                permission.permission_service.add_user_if_not_exists(
                     auth_user.id)
 
+        # Store user info in request.state for access by GET endpoints
+        if auth_user is not None:
+            request.state.auth_user = auth_user
+        else:
+            request.state.auth_user = None
+
         body = await request.body()
         if auth_user and body:
             try:
@@ -177,6 +188,12 @@ class AuthProxyMiddleware(starlette.middleware.base.BaseHTTPMiddleware):
                             f'"env_vars" in request body is not a dictionary '
                             f'for request {request.state.request_id}. '
                             'Skipping user info injection into body.')
+                else:
+                    original_json['env_vars'] = {}
+                    original_json['env_vars'][
+                        constants.USER_ID_ENV_VAR] = auth_user.id
+                    original_json['env_vars'][
+                        constants.USER_ENV_VAR] = auth_user.name
                 request._body = json.dumps(original_json).encode('utf-8')  # pylint: disable=protected-access
         return await call_next(request)
 
@@ -676,6 +693,7 @@ async def launch(launch_body: payloads.LaunchBody,
         func=execution.launch,
         schedule_type=requests_lib.ScheduleType.LONG,
         request_cluster_name=launch_body.cluster_name,
+        retryable=launch_body.retry_until_up,
     )
 
 
diff --git a/sky/skylet/constants.py b/sky/skylet/constants.py
index cc56c05f0..297edc163 100644
--- a/sky/skylet/constants.py
+++ b/sky/skylet/constants.py
@@ -419,3 +419,6 @@ ALL_CLOUDS = ('aws', 'azure', 'gcp', 'ibm', 'lambda', 'scp', 'oci',
               'kubernetes', 'runpod', 'vast', 'vsphere', 'cudo', 'fluidstack',
               'paperspace', 'do', 'nebius', 'ssh')
 # END constants used for service catalog.
+
+# The user ID of the SkyPilot system.
+SKYPILOT_SYSTEM_USER_ID = 'skypilot-system'
diff --git a/sky/skylet/job_lib.py b/sky/skylet/job_lib.py
index 1443d08bb..88e641eca 100644
--- a/sky/skylet/job_lib.py
+++ b/sky/skylet/job_lib.py
@@ -794,7 +794,8 @@ def load_job_queue(payload: str) -> List[Dict[str, Any]]:
     for job in jobs:
         job['status'] = JobStatus(job['status'])
         job['user_hash'] = job['username']
-        job['username'] = global_user_state.get_user(job['user_hash']).name
+        user = global_user_state.get_user(job['user_hash'])
+        job['username'] = user.name if user is not None else None
     return jobs
 
 
diff --git a/sky/skypilot_config.py b/sky/skypilot_config.py
index 4275d07e9..9e3ced448 100644
--- a/sky/skypilot_config.py
+++ b/sky/skypilot_config.py
@@ -167,7 +167,10 @@ def _get_loaded_config_path() -> List[Optional[str]]:
     serialized = _get_config_context().config_path
     if not serialized:
         return []
-    return json.loads(serialized)
+    config_paths = json.loads(serialized)
+    if config_paths is None:
+        return []
+    return config_paths
 
 
 def _set_loaded_config_path(
@@ -762,6 +765,15 @@ def update_api_server_config_no_lock(config: config_utils.Config) -> None:
     Args:
         config: The config to save and sync.
     """
+
+    def is_running_pytest() -> bool:
+        return 'PYTEST_CURRENT_TEST' in os.environ
+
+    # Only allow this function to be called by the API Server in production.
+    if not is_running_pytest() and os.environ.get(
+            constants.ENV_VAR_IS_SKYPILOT_SERVER) is None:
+        raise ValueError('This function can only be called by the API Server.')
+
     global_config_path = _resolve_server_config_path()
     if global_config_path is None:
         global_config_path = get_user_config_path()
diff --git a/sky/templates/jobs-controller.yaml.j2 b/sky/templates/jobs-controller.yaml.j2
index e235c6b04..7fc530958 100644
--- a/sky/templates/jobs-controller.yaml.j2
+++ b/sky/templates/jobs-controller.yaml.j2
@@ -3,6 +3,7 @@
 name: {{dag_name}}
 
 file_mounts:
+  {{remote_original_user_yaml_path}}: {{original_user_dag_path}}
   {{remote_user_yaml_path}}: {{user_yaml_path}}
   {%- if local_user_config_path is not none %}
   {{remote_user_config_path}}: {{local_user_config_path}}
@@ -28,7 +29,7 @@ setup: |
   grep -q 'export SKYPILOT_DEV=' ~/.bashrc || echo 'export SKYPILOT_DEV=1' >> ~/.bashrc
   grep -q 'alias sky-env=' ~/.bashrc || echo 'alias sky-env="{{ sky_activate_python_env }}"' >> ~/.bashrc
   {% endif %}
-
+  
   # Create systemd service file
   mkdir -p ~/.config/systemd/user/
 
@@ -65,6 +66,7 @@ run: |
   # CloudVmRayBackend._exec_code_on_head() calls
   # managed_job_codegen.set_pending() before we get here.
   python -u -m sky.jobs.scheduler {{remote_user_yaml_path}} \
+    --user-yaml-path {{remote_original_user_yaml_path}} \
     --job-id $SKYPILOT_INTERNAL_JOB_ID \
     --env-file {{remote_env_file_path}} \
     --priority {{priority}}
diff --git a/sky/users/model.conf b/sky/users/model.conf
index ecaaa4ad6..414932a1e 100644
--- a/sky/users/model.conf
+++ b/sky/users/model.conf
@@ -12,4 +12,4 @@ g = _, _
 e = some(where (p.eft == allow))
 
 [matchers]
-m = g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act
+m = (g(r.sub, p.sub)|| p.sub == '*') && r.obj == p.obj && r.act == p.act
diff --git a/sky/users/permission.py b/sky/users/permission.py
index 7e7c2b7d9..92e09c9db 100644
--- a/sky/users/permission.py
+++ b/sky/users/permission.py
@@ -3,7 +3,7 @@ import contextlib
 import logging
 import os
 import threading
-from typing import List
+from typing import Generator, List
 
 import casbin
 import filelock
@@ -11,8 +11,11 @@ import sqlalchemy_adapter
 
 from sky import global_user_state
 from sky import sky_logging
+from sky.skylet import constants
 from sky.users import rbac
 
+logging.getLogger('casbin.policy').setLevel(sky_logging.ERROR)
+logging.getLogger('casbin.role').setLevel(sky_logging.ERROR)
 logger = sky_logging.init_logger(__name__)
 
 # Filelocks for the policy update.
@@ -38,17 +41,19 @@ class PermissionService:
                     model_path = os.path.join(os.path.dirname(__file__),
                                               'model.conf')
                     enforcer = casbin.Enforcer(model_path, adapter)
-                    logging.getLogger('casbin.policy').setLevel(
-                        sky_logging.ERROR)
-                    logging.getLogger('casbin.role').setLevel(sky_logging.ERROR)
                     self.enforcer = enforcer
         else:
             self.enforcer = _enforcer_instance.enforcer
-        self._maybe_initialize_policies()
+        with _policy_lock():
+            self._maybe_initialize_policies()
 
-    def _maybe_initialize_policies(self):
+    def _maybe_initialize_policies(self) -> None:
         """Initialize policies if they don't already exist."""
+        # TODO(zhwu): we should avoid running this on client side.
         logger.debug(f'Initializing policies in process: {os.getpid()}')
+        self._load_policy_no_lock()
+
+        policy_updated = False
 
         # Check if policies are already initialized by looking for existing
         # permission policies in the enforcer
@@ -66,6 +71,17 @@ class PermissionService:
                     expected_policies.append(
                         [role, item['path'], item['method']])
 
+        # Add workspace policy
+        workspace_policy_permissions = rbac.get_workspace_policy_permissions()
+        logger.debug(f'Workspace policy permissions from config: '
+                     f'{workspace_policy_permissions}')
+
+        for workspace_name, users in workspace_policy_permissions.items():
+            for user in users:
+                expected_policies.append([user, workspace_name, '*'])
+                logger.debug(f'Expected workspace policy: user={user}, '
+                             f'workspace={workspace_name}')
+
         # Check if all expected policies already exist
         policies_exist = all(
             any(policy == expected
@@ -86,48 +102,71 @@ class PermissionService:
                     for item in blocklist:
                         path = item['path']
                         method = item['method']
+                        logger.debug(f'Adding role policy: role={role}, '
+                                     f'path={path}, method={method}')
                         self.enforcer.add_policy(role, path, method)
-            self.enforcer.save_policy()
+                        policy_updated = True
+
+            for workspace_name, users in workspace_policy_permissions.items():
+                for user in users:
+                    logger.debug(f'Initializing workspace policy: user={user}, '
+                                 f'workspace={workspace_name}')
+                    self.enforcer.add_policy(user, workspace_name, '*')
+                    policy_updated = True
+            logger.debug('Policies initialized successfully')
         else:
             logger.debug('Policies already exist, skipping initialization')
 
         # Always ensure users have default roles (this is idempotent)
         all_users = global_user_state.get_all_users()
-        for user in all_users:
-            self.add_user_if_not_exists(user.id)
+        for existing_user in all_users:
+            user_added = self._add_user_if_not_exists_no_lock(existing_user.id)
+            policy_updated = policy_updated or user_added
+
+        if policy_updated:
+            self.enforcer.save_policy()
 
-    def add_user_if_not_exists(self, user: str) -> None:
+    def add_user_if_not_exists(self, user_id: str) -> None:
         """Add user role relationship."""
         with _policy_lock():
-            user_roles = self.enforcer.get_roles_for_user(user)
-            if not user_roles:
-                logger.info(f'User {user} has no roles, adding'
-                            f' default role {rbac.get_default_role()}')
-                self.enforcer.add_grouping_policy(user, rbac.get_default_role())
-                self.enforcer.save_policy()
-
-    def update_role(self, user: str, new_role: str):
+            self._add_user_if_not_exists_no_lock(user_id)
+
+    def _add_user_if_not_exists_no_lock(self, user_id: str) -> bool:
+        """Add user role relationship without lock.
+
+        Returns:
+            True if the user was added, False otherwise.
+        """
+        user_roles = self.enforcer.get_roles_for_user(user_id)
+        if not user_roles:
+            logger.info(f'User {user_id} has no roles, adding'
+                        f' default role {rbac.get_default_role()}')
+            self.enforcer.add_grouping_policy(user_id, rbac.get_default_role())
+            return True
+        return False
+
+    def update_role(self, user_id: str, new_role: str) -> None:
         """Update user role relationship."""
         with _policy_lock():
             # Get current roles
             self._load_policy_no_lock()
             # Avoid calling get_user_roles, as it will require the lock.
-            current_roles = self.enforcer.get_roles_for_user(user)
+            current_roles = self.enforcer.get_roles_for_user(user_id)
             if not current_roles:
-                logger.warning(f'User {user} has no roles')
+                logger.warning(f'User {user_id} has no roles')
             else:
                 # TODO(hailong): how to handle multiple roles?
                 current_role = current_roles[0]
                 if current_role == new_role:
-                    logger.info(f'User {user} already has role {new_role}')
+                    logger.info(f'User {user_id} already has role {new_role}')
                     return
-                self.enforcer.remove_grouping_policy(user, current_role)
+                self.enforcer.remove_grouping_policy(user_id, current_role)
 
             # Update user role
-            self.enforcer.add_grouping_policy(user, new_role)
+            self.enforcer.add_grouping_policy(user_id, new_role)
             self.enforcer.save_policy()
 
-    def get_user_roles(self, user: str) -> List[str]:
+    def get_user_roles(self, user_id: str) -> List[str]:
         """Get all roles for a user.
 
         This method returns all roles that the user has, including inherited
@@ -140,10 +179,11 @@ class PermissionService:
         Returns:
             A list of role names that the user has.
         """
-        self._load_policy()
-        return self.enforcer.get_roles_for_user(user)
+        self._load_policy_no_lock()
+        return self.enforcer.get_roles_for_user(user_id)
 
-    def check_permission(self, user: str, path: str, method: str) -> bool:
+    def check_endpoint_permission(self, user_id: str, path: str,
+                                  method: str) -> bool:
         """Check permission."""
         # We intentionally don't load the policy here, as it is a hot path, and
         # we don't support updating the policy.
@@ -151,28 +191,105 @@ class PermissionService:
         # it is a hot path in every request. It is ok to have a stale policy,
         # as long as it is eventually consistent.
         # self._load_policy_no_lock()
-        return self.enforcer.enforce(user, path, method)
+        return self.enforcer.enforce(user_id, path, method)
 
     def _load_policy_no_lock(self):
         """Load policy from storage."""
         self.enforcer.load_policy()
 
-    def _load_policy(self):
+    def load_policy(self):
         """Load policy from storage with lock."""
         with _policy_lock():
             self._load_policy_no_lock()
 
+    def check_workspace_permission(self, user_id: str,
+                                   workspace_name: str) -> bool:
+        """Check workspace permission.
+
+        This method checks if a user has permission to access a specific
+        workspace.
+
+        For private workspaces, the user must have explicit permission.
+
+        For public workspaces, the permission is granted via a wildcard policy
+        ('*').
+        """
+        if os.getenv(constants.ENV_VAR_IS_SKYPILOT_SERVER) is None:
+            # When it is not on API server, we allow all users to access all
+            # workspaces, as the workspace check has been done on API server.
+            return True
+        role = self.get_user_roles(user_id)
+        if rbac.RoleName.ADMIN.value in role:
+            return True
+        # The Casbin model matcher already handles the wildcard '*' case:
+        # m = (g(r.sub, p.sub)|| p.sub == '*') && r.obj == p.obj &&
+        # r.act == p.act
+        # This means if there's a policy ('*', workspace_name, '*'), it will
+        # match any user
+        result = self.enforcer.enforce(user_id, workspace_name, '*')
+        logger.debug(f'Workspace permission check: user={user_id}, '
+                     f'workspace={workspace_name}, result={result}')
+        return result
+
+    def add_workspace_policy(self, workspace_name: str,
+                             users: List[str]) -> None:
+        """Add workspace policy.
+
+        Args:
+            workspace_name: Name of the workspace
+            users: List of user IDs that should have access.
+                   For public workspaces, this should be ['*'].
+                   For private workspaces, this should be specific user IDs.
+        """
+        with _policy_lock():
+            for user in users:
+                logger.debug(f'Adding workspace policy: user={user}, '
+                             f'workspace={workspace_name}')
+                self.enforcer.add_policy(user, workspace_name, '*')
+            self.enforcer.save_policy()
+
+    def update_workspace_policy(self, workspace_name: str,
+                                users: List[str]) -> None:
+        """Update workspace policy.
+
+        Args:
+            workspace_name: Name of the workspace
+            users: List of user IDs that should have access.
+                   For public workspaces, this should be ['*'].
+                   For private workspaces, this should be specific user IDs.
+        """
+        with _policy_lock():
+            self._load_policy_no_lock()
+            # Remove all existing policies for this workspace
+            self.enforcer.remove_filtered_policy(1, workspace_name)
+            # Add new policies
+            for user in users:
+                logger.debug(f'Updating workspace policy: user={user}, '
+                             f'workspace={workspace_name}')
+                self.enforcer.add_policy(user, workspace_name, '*')
+            self.enforcer.save_policy()
+
+    def remove_workspace_policy(self, workspace_name: str) -> None:
+        """Remove workspace policy."""
+        with _policy_lock():
+            self.enforcer.remove_filtered_policy(1, workspace_name)
+            self.enforcer.save_policy()
+
 
 @contextlib.contextmanager
-def _policy_lock():
+def _policy_lock() -> Generator[None, None, None]:
     """Context manager for policy update lock."""
     try:
         with filelock.FileLock(POLICY_UPDATE_LOCK_PATH,
                                POLICY_UPDATE_LOCK_TIMEOUT_SECONDS):
             yield
     except filelock.Timeout as e:
-        raise RuntimeError(f'Failed to load policy due to a timeout '
+        raise RuntimeError(f'Failed to reload policy due to a timeout '
                            f'when trying to acquire the lock at '
                            f'{POLICY_UPDATE_LOCK_PATH}. '
                            'Please try again or manually remove the lock '
                            f'file if you believe it is stale.') from e
+
+
+# Singleton instance of PermissionService for other modules to use.
+permission_service = PermissionService()
diff --git a/sky/users/rbac.py b/sky/users/rbac.py
index e16231ded..f78bb046d 100644
--- a/sky/users/rbac.py
+++ b/sky/users/rbac.py
@@ -5,6 +5,8 @@ from typing import Dict, List
 
 from sky import sky_logging
 from sky import skypilot_config
+from sky.skylet import constants
+from sky.workspaces import utils as workspaces_utils
 
 logger = sky_logging.init_logger(__name__)
 
@@ -84,3 +86,27 @@ def get_role_permissions(
             }
         }
     return config_permissions
+
+
+def get_workspace_policy_permissions() -> Dict[str, List[str]]:
+    """Get workspace policy permissions from config.
+
+    Returns:
+        A dictionary of workspace policy permissions.
+        Example:
+        {
+            'workspace1': ['user1-id', 'user2-id'],
+            'workspace2': ['user3-id', 'user4-id']
+            'default': ['*']
+        }
+    """
+    current_workspaces = skypilot_config.get_nested(('workspaces',),
+                                                    default_value={})
+    if constants.SKYPILOT_DEFAULT_WORKSPACE not in current_workspaces:
+        current_workspaces[constants.SKYPILOT_DEFAULT_WORKSPACE] = {}
+    workspaces_to_policy = {}
+    for workspace_name, workspace_config in current_workspaces.items():
+        users = workspaces_utils.get_workspace_users(workspace_config)
+        workspaces_to_policy[workspace_name] = users
+    logger.debug(f'Workspace policy permissions: {workspaces_to_policy}')
+    return workspaces_to_policy
diff --git a/sky/users/server.py b/sky/users/server.py
index 65d264acc..6ce305eab 100644
--- a/sky/users/server.py
+++ b/sky/users/server.py
@@ -1,6 +1,5 @@
 """REST API for workspace management."""
 
-import hashlib
 from typing import Any, Dict, List
 
 import fastapi
@@ -8,16 +7,15 @@ import fastapi
 from sky import global_user_state
 from sky import sky_logging
 from sky.server.requests import payloads
+from sky.skylet import constants
 from sky.users import permission
 from sky.users import rbac
-from sky.utils import common_utils
+from sky.utils import common
 
 logger = sky_logging.init_logger(__name__)
 
 router = fastapi.APIRouter()
 
-permission_service = permission.PermissionService()
-
 
 @router.get('')
 async def users() -> List[Dict[str, Any]]:
@@ -25,7 +23,7 @@ async def users() -> List[Dict[str, Any]]:
     all_users = []
     user_list = global_user_state.get_all_users()
     for user in user_list:
-        user_roles = permission_service.get_user_roles(user.id)
+        user_roles = permission.permission_service.get_user_roles(user.id)
         all_users.append({
             'id': user.id,
             'name': user.name,
@@ -39,13 +37,11 @@ async def get_current_user_role(request: fastapi.Request):
     """Get current user's role."""
     # TODO(hailong): is there a reliable way to get the user
     # hash for the request without 'X-Auth-Request-Email' header?
-    if 'X-Auth-Request-Email' not in request.headers:
+    auth_user = request.state.auth_user
+    if auth_user is None:
         return {'name': '', 'role': rbac.RoleName.ADMIN.value}
-    user_name = request.headers['X-Auth-Request-Email']
-    user_hash = hashlib.md5(
-        user_name.encode()).hexdigest()[:common_utils.USER_HASH_LENGTH]
-    user_roles = permission_service.get_user_roles(user_hash)
-    return {'name': user_name, 'role': user_roles[0] if user_roles else ''}
+    user_roles = permission.permission_service.get_user_roles(auth_user.id)
+    return {'name': auth_user.name, 'role': user_roles[0] if user_roles else ''}
 
 
 @router.post('/update')
@@ -58,9 +54,14 @@ async def user_update(user_update_body: payloads.UserUpdateBody) -> None:
         raise fastapi.HTTPException(status_code=400,
                                     detail=f'Invalid role: {role}')
     user_info = global_user_state.get_user(user_id)
-    if not user_info.name:
+    if user_info is None:
         raise fastapi.HTTPException(status_code=400,
                                     detail=f'User {user_id} does not exist')
+    # Disallow updating roles for the internal users.
+    if user_info.id in [common.SERVER_ID, constants.SKYPILOT_SYSTEM_USER_ID]:
+        raise fastapi.HTTPException(status_code=400,
+                                    detail=f'Cannot update role for internal '
+                                    f'API server user {user_info.name}')
 
     # Update user role in casbin policy
-    permission_service.update_role(user_id, role)
+    permission.permission_service.update_role(user_info.id, role)
diff --git a/sky/utils/common.py b/sky/utils/common.py
index 95a25d611..34fbbc9ea 100644
--- a/sky/utils/common.py
+++ b/sky/utils/common.py
@@ -5,6 +5,7 @@ import enum
 import os
 from typing import Generator
 
+from sky import models
 from sky.skylet import constants
 from sky.utils import common_utils
 
@@ -25,10 +26,13 @@ JOB_CONTROLLER_NAME: str = f'{JOB_CONTROLLER_PREFIX}{SERVER_ID}'
 
 
 @contextlib.contextmanager
-def with_server_user_hash() -> Generator[None, None, None]:
+def with_server_user() -> Generator[None, None, None]:
     """Temporarily set the user hash to common.SERVER_ID."""
     old_env_user_hash = os.getenv(constants.USER_ID_ENV_VAR)
+    # TODO(zhwu): once we have fully moved our code to use `get_current_user()`
+    # instead of `common_utils.get_user_hash()`, we can remove the env override.
     os.environ[constants.USER_ID_ENV_VAR] = SERVER_ID
+    common_utils.set_current_user(models.User.get_current_user())
     try:
         yield
     finally:
@@ -36,6 +40,7 @@ def with_server_user_hash() -> Generator[None, None, None]:
             os.environ[constants.USER_ID_ENV_VAR] = old_env_user_hash
         else:
             os.environ.pop(constants.USER_ID_ENV_VAR)
+        common_utils.set_current_user(models.User.get_current_user())
 
 
 class StatusRefreshMode(enum.Enum):
diff --git a/sky/utils/common_utils.py b/sky/utils/common_utils.py
index 636d86456..a0ca0e97a 100644
--- a/sky/utils/common_utils.py
+++ b/sky/utils/common_utils.py
@@ -20,6 +20,7 @@ import uuid
 import jsonschema
 
 from sky import exceptions
+from sky import models
 from sky import sky_logging
 from sky.adaptors import common as adaptors_common
 from sky.skylet import constants
@@ -256,11 +257,13 @@ class Backoff:
 _current_command: Optional[str] = None
 _current_client_entrypoint: Optional[str] = None
 _using_remote_api_server: Optional[bool] = None
+_current_user: Optional['models.User'] = None
 
 
-def set_client_status(client_entrypoint: Optional[str],
-                      client_command: Optional[str],
-                      using_remote_api_server: bool):
+def set_request_context(client_entrypoint: Optional[str],
+                        client_command: Optional[str],
+                        using_remote_api_server: bool,
+                        user: Optional['models.User']):
     """Override the current client entrypoint and command.
 
     This is useful when we are on the SkyPilot API server side and we have a
@@ -269,9 +272,11 @@ def set_client_status(client_entrypoint: Optional[str],
     global _current_command
     global _current_client_entrypoint
     global _using_remote_api_server
+    global _current_user
     _current_command = client_command
     _current_client_entrypoint = client_entrypoint
     _using_remote_api_server = using_remote_api_server
+    _current_user = user
 
 
 def get_current_command() -> str:
@@ -286,6 +291,19 @@ def get_current_command() -> str:
     return get_pretty_entrypoint_cmd()
 
 
+def get_current_user() -> 'models.User':
+    """Returns the current user."""
+    if _current_user is not None:
+        return _current_user
+    return models.User.get_current_user()
+
+
+def set_current_user(user: 'models.User'):
+    """Sets the current user."""
+    global _current_user
+    _current_user = user
+
+
 def get_current_client_entrypoint(server_entrypoint: str) -> str:
     """Returns the current client entrypoint.
 
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index 3ee265a29..dcf14e1e4 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -1249,6 +1249,15 @@ def get_config_schema():
             'properties': {
                 # Explicit definition for GCP allows both project_id and
                 # disabled
+                'private': {
+                    'type': 'boolean',
+                },
+                'allowed_users': {
+                    'type': 'array',
+                    'items': {
+                        'type': 'string',
+                    },
+                },
                 'gcp': {
                     'type': 'object',
                     'properties': {
diff --git a/sky/workspaces/core.py b/sky/workspaces/core.py
index 35ba912ea..20ffde62f 100644
--- a/sky/workspaces/core.py
+++ b/sky/workspaces/core.py
@@ -1,20 +1,24 @@
 """Workspace management core."""
 
 import concurrent.futures
-from typing import Any, Callable, Dict
+from typing import Any, Callable, Dict, List
 
 import filelock
 
 from sky import check as sky_check
 from sky import exceptions
 from sky import global_user_state
+from sky import models
 from sky import sky_logging
 from sky import skypilot_config
 from sky.skylet import constants
 from sky.usage import usage_lib
+from sky.users import permission
+from sky.utils import annotations
 from sky.utils import common_utils
 from sky.utils import config_utils
 from sky.utils import schemas
+from sky.workspaces import utils as workspaces_utils
 
 logger = sky_logging.init_logger(__name__)
 
@@ -28,10 +32,7 @@ _WORKSPACE_CONFIG_LOCK_TIMEOUT_SECONDS = 60
 
 def get_workspaces() -> Dict[str, Any]:
     """Returns the workspace config."""
-    workspaces = skypilot_config.get_nested(('workspaces',), default_value={})
-    if constants.SKYPILOT_DEFAULT_WORKSPACE not in workspaces:
-        workspaces[constants.SKYPILOT_DEFAULT_WORKSPACE] = {}
-    return workspaces
+    return workspaces_for_user(common_utils.get_current_user().id)
 
 
 def _update_workspaces_config(
@@ -160,7 +161,7 @@ def _check_workspaces_have_no_active_resources(
                 f'{len(workspace_clusters)} active cluster(s): {cluster_list}')
 
         if workspace_active_jobs:
-            job_names = [job['job_id'] for job in workspace_active_jobs]
+            job_names = [str(job['job_id']) for job in workspace_active_jobs]
             job_list = ', '.join(job_names)
             workspace_errors.append(
                 f'{len(workspace_active_jobs)} active managed job(s): '
@@ -223,14 +224,19 @@ def update_workspace(workspace_name: str, config: Dict[str,
         FileNotFoundError: If the config file cannot be found.
         PermissionError: If the config file cannot be written.
     """
+    _validate_workspace_config(workspace_name, config)
+
     # Check for active clusters and managed jobs in the workspace
+    # TODO(zhwu): we should allow the edits that only contain changes to
+    # allowed_users or private.
     _check_workspace_has_no_active_resources(workspace_name, 'update')
 
-    _validate_workspace_config(workspace_name, config)
-
     def update_workspace_fn(workspaces: Dict[str, Any]) -> None:
         """Function to update workspace inside the lock."""
         workspaces[workspace_name] = config
+        users = workspaces_utils.get_workspace_users(config)
+        permission_service = permission.permission_service
+        permission_service.update_workspace_policy(workspace_name, users)
 
     # Use the internal helper function to save
     result = _update_workspaces_config(update_workspace_fn)
@@ -275,6 +281,10 @@ def create_workspace(workspace_name: str, config: Dict[str,
             raise ValueError(f'Workspace {workspace_name!r} already exists. '
                              'Use update instead.')
         workspaces[workspace_name] = config
+        # Add policy for the workspace and allowed users
+        users = workspaces_utils.get_workspace_users(config)
+        permission_service = permission.permission_service
+        permission_service.add_workspace_policy(workspace_name, users)
 
     # Use the internal helper function to save
     result = _update_workspaces_config(create_workspace_fn)
@@ -324,6 +334,8 @@ def delete_workspace(workspace_name: str) -> Dict[str, Any]:
         if workspace_name not in workspaces:
             raise ValueError(f'Workspace {workspace_name!r} does not exist.')
         del workspaces[workspace_name]
+        permission_service = permission.permission_service
+        permission_service.remove_workspace_policy(workspace_name)
 
     # Use the internal helper function to save
     return _update_workspaces_config(delete_workspace_fn)
@@ -369,6 +381,9 @@ def update_config(config: Dict[str, Any]) -> Dict[str, Any]:
 
     # Check for API server changes and validate them
     current_config = skypilot_config.to_dict()
+    # If there is no changes to the config, we can return early
+    if current_config == config:
+        return config
 
     current_endpoint = current_config.get('api_server', {}).get('endpoint')
     new_endpoint = config.get('api_server', {}).get('endpoint')
@@ -382,15 +397,27 @@ def update_config(config: Dict[str, Any]) -> Dict[str, Any]:
 
     # Collect all workspaces that need to be checked for active resources
     workspaces_to_check = []
+    workspaces_to_check_policy: Dict[str, Dict[str, List[str]]] = {
+        'add': {},
+        'update': {},
+        'delete': {}
+    }
 
     # Check each workspace that is being modified
     for workspace_name, new_workspace_config in new_workspaces.items():
+        if workspace_name not in current_workspaces:
+            users = workspaces_utils.get_workspace_users(new_workspace_config)
+            workspaces_to_check_policy['add'][workspace_name] = users
+            continue
+
         current_workspace_config = current_workspaces.get(workspace_name, {})
 
         # If workspace configuration is changing, validate and mark for checking
         if current_workspace_config != new_workspace_config:
             _validate_workspace_config(workspace_name, new_workspace_config)
             workspaces_to_check.append((workspace_name, 'update'))
+            users = workspaces_utils.get_workspace_users(new_workspace_config)
+            workspaces_to_check_policy['update'][workspace_name] = users
 
     # Check for workspace deletions
     for workspace_name in current_workspaces:
@@ -400,6 +427,7 @@ def update_config(config: Dict[str, Any]) -> Dict[str, Any]:
                 raise ValueError(f'Cannot delete the default workspace '
                                  f'{constants.SKYPILOT_DEFAULT_WORKSPACE!r}.')
             workspaces_to_check.append((workspace_name, 'delete'))
+            workspaces_to_check_policy['delete'][workspace_name] = ['*']
 
     # Check all workspaces for active resources in one efficient call
     _check_workspaces_have_no_active_resources(workspaces_to_check)
@@ -412,6 +440,18 @@ def update_config(config: Dict[str, Any]) -> Dict[str, Any]:
             # Convert to config_utils.Config and save
             config_obj = config_utils.Config.from_dict(config)
             skypilot_config.update_api_server_config_no_lock(config_obj)
+            permission_service = permission.permission_service
+            for operation, workspaces in workspaces_to_check_policy.items():
+                for workspace_name, users in workspaces.items():
+                    if operation == 'add':
+                        permission_service.add_workspace_policy(
+                            workspace_name, users)
+                    elif operation == 'update':
+                        permission_service.update_workspace_policy(
+                            workspace_name, users)
+                    elif operation == 'delete':
+                        permission_service.remove_workspace_policy(
+                            workspace_name)
     except filelock.Timeout as e:
         raise RuntimeError(
             f'Failed to update configuration due to a timeout '
@@ -429,3 +469,55 @@ def update_config(config: Dict[str, Any]) -> Dict[str, Any]:
         # Don't fail the update if the check fails, just warn
 
     return config
+
+
+def reject_request_for_unauthorized_workspace(user: models.User) -> None:
+    """Rejects a request that has no permission to access active workspace.
+
+    Args:
+        user: The user making the request.
+
+    Raises:
+        PermissionDeniedError: If the user does not have permission to access
+            the active workspace.
+    """
+    active_workspace = skypilot_config.get_active_workspace()
+    if not permission.permission_service.check_workspace_permission(
+            user.id, active_workspace):
+        raise exceptions.PermissionDeniedError(
+            f'User {user.name} ({user.id}) does not have '
+            f'permission to access workspace {active_workspace!r}')
+
+
+def is_workspace_private(workspace_config: Dict[str, Any]) -> bool:
+    """Check if a workspace is private.
+
+    Args:
+        workspace_config: The workspace configuration dictionary.
+
+    Returns:
+        True if the workspace is private, False if it's public.
+    """
+    return workspace_config.get('private', False)
+
+
+@annotations.lru_cache(scope='request', maxsize=1)
+def workspaces_for_user(user_id: str) -> Dict[str, Any]:
+    """Returns the workspaces that the user has access to.
+
+    Args:
+        user_id: The user id to check.
+
+    Returns:
+        A map from workspace name to workspace configuration.
+    """
+    workspaces = skypilot_config.get_nested(('workspaces',), default_value={})
+    if constants.SKYPILOT_DEFAULT_WORKSPACE not in workspaces:
+        workspaces[constants.SKYPILOT_DEFAULT_WORKSPACE] = {}
+    user_workspaces = {}
+
+    for workspace_name, workspace_config in workspaces.items():
+        if permission.permission_service.check_workspace_permission(
+                user_id, workspace_name):
+            user_workspaces[workspace_name] = workspace_config
+    return user_workspaces
diff --git a/sky/workspaces/server.py b/sky/workspaces/server.py
index 3adfc860c..5afa1511b 100644
--- a/sky/workspaces/server.py
+++ b/sky/workspaces/server.py
@@ -14,10 +14,18 @@ router = fastapi.APIRouter()
 # pylint: disable=redefined-builtin
 async def get(request: fastapi.Request) -> None:
     """Gets workspace config on the server."""
+    # Have to manually inject user info into the request body because the
+    # request body is not available in the GET endpoint.
+    auth_user = request.state.auth_user
+    auth_user_env_vars_kwargs = {
+        'env_vars': auth_user.to_env_vars()
+    } if auth_user else {}
+    request_body = payloads.RequestBody(**auth_user_env_vars_kwargs)
+
     executor.schedule_request(
         request_id=request.state.request_id,
         request_name='workspaces.get',
-        request_body=payloads.RequestBody(),
+        request_body=request_body,
         func=core.get_workspaces,
         schedule_type=api_requests.ScheduleType.SHORT,
     )
@@ -65,10 +73,15 @@ async def delete(request: fastapi.Request,
 @router.get('/config')
 async def get_config(request: fastapi.Request) -> None:
     """Gets the entire SkyPilot configuration."""
+    auth_user = request.state.auth_user
+    auth_user_env_vars_kwargs = {
+        'env_vars': auth_user.to_env_vars()
+    } if auth_user else {}
+    get_config_body = payloads.GetConfigBody(**auth_user_env_vars_kwargs)
     executor.schedule_request(
         request_id=request.state.request_id,
         request_name='workspaces.get_config',
-        request_body=payloads.GetConfigBody(),
+        request_body=get_config_body,
         func=core.get_config,
         schedule_type=api_requests.ScheduleType.SHORT,
     )
diff --git a/sky/workspaces/utils.py b/sky/workspaces/utils.py
new file mode 100644
index 000000000..aa2aeaf2a
--- /dev/null
+++ b/sky/workspaces/utils.py
@@ -0,0 +1,56 @@
+"""Utils for workspaces."""
+import collections
+from typing import Any, Dict, List
+
+from sky import global_user_state
+from sky import sky_logging
+
+logger = sky_logging.init_logger(__name__)
+
+
+def get_workspace_users(workspace_config: Dict[str, Any]) -> List[str]:
+    """Get the users that should have access to a workspace.
+
+    workspace_config is a dict with the following keys:
+    - private: bool
+    - allowed_users: list of user names or IDs
+
+    This function will automatically resolve the user names to IDs.
+
+    Args:
+        workspace_config: The configuration of the workspace.
+
+    Returns:
+        List of user IDs that should have access to the workspace.
+        For private workspaces, returns specific user IDs.
+        For public workspaces, returns ['*'] to indicate all users.
+    """
+    if workspace_config.get('private', False):
+        user_ids = []
+        workspace_user_name_or_ids = workspace_config.get('allowed_users', [])
+        all_users = global_user_state.get_all_users()
+        all_user_ids = {user.id for user in all_users}
+        all_user_map = collections.defaultdict(list)
+        for user in all_users:
+            all_user_map[user.name].append(user.id)
+
+        # Resolve user names to IDs
+        for user_name_or_id in workspace_user_name_or_ids:
+            if user_name_or_id in all_user_ids:
+                user_ids.append(user_name_or_id)
+            elif user_name_or_id in all_user_map:
+                if len(all_user_map[user_name_or_id]) > 1:
+                    user_ids_str = ', '.join(all_user_map[user_name_or_id])
+                    raise ValueError(
+                        f'User {user_name_or_id!r} has multiple IDs: '
+                        f'{user_ids_str}. Please specify the user '
+                        f'ID instead.')
+                user_ids.append(all_user_map[user_name_or_id][0])
+            else:
+                logger.warning(
+                    f'User {user_name_or_id!r} not found in all users')
+                continue
+        return user_ids
+    else:
+        # Public workspace - return '*' to indicate all users should have access
+        return ['*']


Test patch:
diff --git a/tests/common_test_fixtures.py b/tests/common_test_fixtures.py
index ae10ac6b3..486270e91 100644
--- a/tests/common_test_fixtures.py
+++ b/tests/common_test_fixtures.py
@@ -357,8 +357,8 @@ def mock_queue(monkeypatch):
             self.queue_map = {}
 
         def put(self, item):
-            # Add to the map; item is assumed to be a tuple (request_id, ignore_return_value)
-            request_id, ignore_return_value = item
+            # Add to the map; item is assumed to be a tuple (request_id, ignore_return_value, retryable)
+            request_id, ignore_return_value, _ = item
             self.queue_map[request_id] = ignore_return_value
 
         def get(self, request_id):
diff --git a/tests/smoke_tests/test_workspaces.py b/tests/smoke_tests/test_workspaces.py
index 2e64eac52..fb0cfd1eb 100644
--- a/tests/smoke_tests/test_workspaces.py
+++ b/tests/smoke_tests/test_workspaces.py
@@ -20,7 +20,10 @@ from sky import skypilot_config
                   'workspaces:\n'
                   '  ws-1: {}\n'
                   '  ws-2: {}\n'
-                  'and restart the API server.')
+                  'and restart the API server.'
+                  'The reason for not using env var SKYPILOT_CONFIG to '
+                  'override the global config is that the project-level config '
+                  'will be ignored in that case.')
 def test_workspace_switching(generic_cloud: str):
     # Test switching between workspaces by modifying .sky.yaml.
     #
diff --git a/tests/unit_tests/test_sky/server/test_common.py b/tests/unit_tests/test_sky/server/test_common.py
index 2d0a247cf..31b86ffee 100644
--- a/tests/unit_tests/test_sky/server/test_common.py
+++ b/tests/unit_tests/test_sky/server/test_common.py
@@ -208,7 +208,7 @@ def test_install_server_version_command():
 @pytest.fixture
 def mock_all_dependencies():
     """Mock all dependencies used in reload_for_new_request."""
-    with mock.patch('sky.utils.common_utils.set_client_status') as mock_status, \
+    with mock.patch('sky.utils.common_utils.set_request_context') as mock_status, \
          mock.patch('sky.usage.usage_lib.messages.reset') as mock_reset, \
          mock.patch('sky.sky_logging.reload_logger') as mock_logger:
         yield {
@@ -233,6 +233,7 @@ allowed_clouds:
         client_entrypoint='test_entry',
         client_command='test_cmd',
         using_remote_api_server=False,
+        user=mock.Mock(id='test_user'),
     )
     assert skypilot_config.get_nested(keys=('allowed_clouds',),
                                       default_value=None) == ['aws']
@@ -244,6 +245,7 @@ allowed_clouds:
         client_entrypoint='test_entry',
         client_command='test_cmd',
         using_remote_api_server=False,
+        user=mock.Mock(id='test_user'),
     )
     assert skypilot_config.get_nested(keys=('allowed_clouds',),
                                       default_value=None) == ['gcp']
diff --git a/tests/unit_tests/test_sky/users/__init__.py b/tests/unit_tests/test_sky/users/__init__.py
new file mode 100644
index 000000000..e4afcce75
--- /dev/null
+++ b/tests/unit_tests/test_sky/users/__init__.py
@@ -0,0 +1,1 @@
+# Test package for sky.users module
diff --git a/tests/unit_tests/test_sky/users/test_permission.py b/tests/unit_tests/test_sky/users/test_permission.py
new file mode 100644
index 000000000..1cc0059fb
--- /dev/null
+++ b/tests/unit_tests/test_sky/users/test_permission.py
@@ -0,0 +1,754 @@
+"""Unit tests for permission service using pytest."""
+
+import os
+import threading
+import time
+from unittest import mock
+
+import pytest
+
+from sky import models
+from sky.skylet import constants
+from sky.users import permission
+from sky.users import rbac
+
+
+@pytest.fixture
+def mock_users():
+    """Create mock users for testing."""
+    user1 = models.User(id='user1', name='Alice')
+    user2 = models.User(id='user2', name='Bob')
+    user3 = models.User(id='user3', name='Charlie')
+    return [user1, user2, user3]
+
+
+@pytest.fixture
+def reset_permission_singleton():
+    """Reset permission singleton before and after each test."""
+    # Reset before test
+    permission._enforcer_instance = None
+    yield
+    # Reset after test
+    permission._enforcer_instance = None
+
+
+@pytest.fixture
+def cleanup_env_vars():
+    """Clean up environment variables after each test."""
+    yield
+    env_vars_to_remove = [constants.ENV_VAR_IS_SKYPILOT_SERVER]
+    for env_var in env_vars_to_remove:
+        if env_var in os.environ:
+            del os.environ[env_var]
+
+
+@pytest.mark.usefixtures("reset_permission_singleton", "cleanup_env_vars")
+class TestPermissionService:
+    """Test permission service functionality."""
+
+    @mock.patch('sky.users.permission._policy_lock')
+    @mock.patch('sky.users.permission.sqlalchemy_adapter.Adapter')
+    @mock.patch('sky.users.permission.casbin.Enforcer')
+    @mock.patch('sky.global_user_state.SQLALCHEMY_ENGINE')
+    def test_permission_service_initialization(self, mock_engine,
+                                               mock_enforcer_class,
+                                               mock_adapter_class,
+                                               mock_policy_lock):
+        """Test permission service initialization."""
+        mock_enforcer = mock.Mock()
+        mock_enforcer_class.return_value = mock_enforcer
+        mock_adapter = mock.Mock()
+        mock_adapter_class.return_value = mock_adapter
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        with mock.patch.object(permission.PermissionService,
+                               '_maybe_initialize_policies'):
+            service = permission.PermissionService()
+
+            # Verify SQLAlchemy adapter was created with the correct engine
+            mock_adapter_class.assert_called_once_with(mock_engine)
+
+            # Verify Casbin enforcer was created with correct model path
+            args, kwargs = mock_enforcer_class.call_args
+            assert args[0].endswith('model.conf')
+            assert args[1] == mock_adapter
+
+            # Verify the enforcer is stored
+            assert service.enforcer == mock_enforcer
+
+    @mock.patch('sky.users.permission._policy_lock')
+    @mock.patch('sky.users.rbac.get_workspace_policy_permissions')
+    @mock.patch('sky.users.rbac.get_role_permissions')
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_maybe_initialize_policies_new_setup(self, mock_get_users,
+                                                 mock_get_role_perms,
+                                                 mock_get_workspace_perms,
+                                                 mock_policy_lock, mock_users):
+        """Test policy initialization for a new setup."""
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        # Mock enforcer
+        mock_enforcer = mock.Mock()
+        mock_enforcer.get_policy.return_value = []  # No existing policies
+        mock_enforcer.add_policy.return_value = True
+        mock_enforcer.add_grouping_policy.return_value = True
+        mock_enforcer.get_roles_for_user.return_value = []  # No existing roles
+
+        # Mock role permissions
+        mock_get_role_perms.return_value = {
+            'admin': {
+                'permissions': {
+                    'blocklist': []
+                }
+            },
+            'user': {
+                'permissions': {
+                    'blocklist': [{
+                        'path': '/workspaces/config',
+                        'method': 'POST'
+                    }]
+                }
+            }
+        }
+
+        # Mock workspace permissions
+        mock_get_workspace_perms.return_value = {
+            'default': ['*'],
+            'private-ws': ['user1', 'user2']
+        }
+
+        # Mock users
+        mock_get_users.return_value = mock_users
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+            service._maybe_initialize_policies()
+
+            # Verify policies were added for user role blocklist
+            mock_enforcer.add_policy.assert_any_call('user',
+                                                     '/workspaces/config',
+                                                     'POST')
+
+            # Verify workspace policies were added
+            mock_enforcer.add_policy.assert_any_call('*', 'default', '*')
+            mock_enforcer.add_policy.assert_any_call('user1', 'private-ws', '*')
+            mock_enforcer.add_policy.assert_any_call('user2', 'private-ws', '*')
+
+            # Verify users were assigned default roles
+            for user in mock_users:
+                mock_enforcer.add_grouping_policy.assert_any_call(
+                    user.id, rbac.get_default_role())
+
+            # Verify policy was saved
+            mock_enforcer.save_policy.assert_called()
+
+    @mock.patch('sky.users.permission._policy_lock')
+    @mock.patch('sky.users.rbac.get_workspace_policy_permissions')
+    @mock.patch('sky.users.rbac.get_role_permissions')
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_maybe_initialize_policies_existing_policies(
+            self, mock_get_users, mock_get_role_perms, mock_get_workspace_perms,
+            mock_policy_lock, mock_users):
+        """Test policy initialization when policies already exist (idempotent)."""
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        # Mock enforcer
+        mock_enforcer = mock.Mock()
+        # Simulate existing policies that match expected policies
+        existing_policies = [['user', '/workspaces/config', 'POST'],
+                             ['*', 'default', '*'],
+                             ['user1', 'private-ws', '*'],
+                             ['user2', 'private-ws', '*']]
+        mock_enforcer.get_policy.return_value = existing_policies
+        mock_enforcer.add_policy.return_value = True
+        mock_enforcer.add_grouping_policy.return_value = True
+        mock_enforcer.get_roles_for_user.return_value = [
+            'user'
+        ]  # Users already have roles
+
+        # Mock role permissions
+        mock_get_role_perms.return_value = {
+            'admin': {
+                'permissions': {
+                    'blocklist': []
+                }
+            },
+            'user': {
+                'permissions': {
+                    'blocklist': [{
+                        'path': '/workspaces/config',
+                        'method': 'POST'
+                    }]
+                }
+            }
+        }
+
+        # Mock workspace permissions
+        mock_get_workspace_perms.return_value = {
+            'default': ['*'],
+            'private-ws': ['user1', 'user2']
+        }
+
+        # Mock users
+        mock_get_users.return_value = mock_users
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+            service._maybe_initialize_policies()
+
+            # Verify no new policies were added (since they already exist)
+            mock_enforcer.remove_filtered_policy.assert_not_called()
+            # save_policy should not be called if no updates were made
+            # (users already have roles, policies already exist)
+
+    def test_add_user_if_not_exists_new_user(self):
+        """Test adding a new user that doesn't exist."""
+        mock_enforcer = mock.Mock()
+        mock_enforcer.get_roles_for_user.return_value = []  # No existing roles
+        mock_enforcer.add_grouping_policy.return_value = True
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+
+            result = service._add_user_if_not_exists_no_lock('new_user')
+
+            assert result is True
+            mock_enforcer.add_grouping_policy.assert_called_once_with(
+                'new_user', rbac.get_default_role())
+
+    def test_add_user_if_not_exists_existing_user(self):
+        """Test adding a user that already exists."""
+        mock_enforcer = mock.Mock()
+        mock_enforcer.get_roles_for_user.return_value = [
+            'user'
+        ]  # User already has roles
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+
+            result = service._add_user_if_not_exists_no_lock('existing_user')
+
+            assert result is False
+            mock_enforcer.add_grouping_policy.assert_not_called()
+
+    @mock.patch('sky.users.permission._policy_lock')
+    def test_update_role_new_role(self, mock_policy_lock):
+        """Test updating user role to a new role."""
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        mock_enforcer = mock.Mock()
+        mock_enforcer.get_roles_for_user.return_value = ['user']  # Current role
+        mock_enforcer.remove_grouping_policy.return_value = True
+        mock_enforcer.add_grouping_policy.return_value = True
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+            service._load_policy_no_lock = mock.Mock()
+
+            service.update_role('user1', 'admin')
+
+            # Verify old role was removed and new role was added
+            mock_enforcer.remove_grouping_policy.assert_called_once_with(
+                'user1', 'user')
+            mock_enforcer.add_grouping_policy.assert_called_once_with(
+                'user1', 'admin')
+            mock_enforcer.save_policy.assert_called_once()
+
+    @mock.patch('sky.users.permission._policy_lock')
+    def test_update_role_same_role(self, mock_policy_lock):
+        """Test updating user role to the same role (no-op)."""
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        mock_enforcer = mock.Mock()
+        mock_enforcer.get_roles_for_user.return_value = ['admin'
+                                                        ]  # Current role
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+            service._load_policy_no_lock = mock.Mock()
+
+            service.update_role('user1', 'admin')
+
+            # Verify no changes were made
+            mock_enforcer.remove_grouping_policy.assert_not_called()
+            mock_enforcer.add_grouping_policy.assert_not_called()
+            mock_enforcer.save_policy.assert_not_called()
+
+    def test_get_user_roles(self):
+        """Test getting user roles."""
+        mock_enforcer = mock.Mock()
+        mock_enforcer.get_roles_for_user.return_value = ['admin', 'user']
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+            service._load_policy_no_lock = mock.Mock()
+
+            roles = service.get_user_roles('user1')
+
+            assert roles == ['admin', 'user']
+            mock_enforcer.get_roles_for_user.assert_called_once_with('user1')
+
+    def test_check_endpoint_permission(self):
+        """Test checking endpoint permissions."""
+        mock_enforcer = mock.Mock()
+        mock_enforcer.enforce.return_value = True
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+
+            result = service.check_endpoint_permission('user1', '/api/test',
+                                                       'GET')
+
+            assert result is True
+            mock_enforcer.enforce.assert_called_once_with(
+                'user1', '/api/test', 'GET')
+
+    def test_check_workspace_permission_non_server(self):
+        """Test workspace permission check when not on API server."""
+        # Ensure ENV_VAR_IS_SKYPILOT_SERVER is not set
+        if constants.ENV_VAR_IS_SKYPILOT_SERVER in os.environ:
+            del os.environ[constants.ENV_VAR_IS_SKYPILOT_SERVER]
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+
+            result = service.check_workspace_permission('user1',
+                                                        'test-workspace')
+
+            # Should always return True when not on API server
+            assert result is True
+
+    def test_check_workspace_permission_admin_user(self):
+        """Test workspace permission check for admin user."""
+        os.environ[constants.ENV_VAR_IS_SKYPILOT_SERVER] = 'true'
+
+        mock_enforcer = mock.Mock()
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+            service.get_user_roles = mock.Mock(return_value=['admin'])
+
+            result = service.check_workspace_permission('admin_user',
+                                                        'test-workspace')
+
+            # Admin should always have access
+            assert result is True
+
+    def test_check_workspace_permission_regular_user(self):
+        """Test workspace permission check for regular user."""
+        os.environ[constants.ENV_VAR_IS_SKYPILOT_SERVER] = 'true'
+
+        mock_enforcer = mock.Mock()
+        mock_enforcer.enforce.return_value = True
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+            service.get_user_roles = mock.Mock(return_value=['user'])
+
+            result = service.check_workspace_permission('user1',
+                                                        'test-workspace')
+
+            assert result is True
+            mock_enforcer.enforce.assert_called_once_with(
+                'user1', 'test-workspace', '*')
+
+    @mock.patch('sky.users.permission._policy_lock')
+    def test_add_workspace_policy(self, mock_policy_lock):
+        """Test adding workspace policy."""
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        mock_enforcer = mock.Mock()
+        mock_enforcer.add_policy.return_value = True
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+
+            service.add_workspace_policy('test-workspace', ['user1', 'user2'])
+
+            # Verify policies were added for each user
+            mock_enforcer.add_policy.assert_any_call('user1', 'test-workspace',
+                                                     '*')
+            mock_enforcer.add_policy.assert_any_call('user2', 'test-workspace',
+                                                     '*')
+            mock_enforcer.save_policy.assert_called_once()
+
+    @mock.patch('sky.users.permission.filelock.FileLock')
+    def test_policy_lock_context_manager(self, mock_filelock):
+        """Test the policy lock context manager."""
+        mock_lock = mock.Mock()
+        mock_lock.__enter__ = mock.Mock(return_value=mock.Mock())
+        mock_lock.__exit__ = mock.Mock(return_value=None)
+        mock_filelock.return_value = mock_lock
+
+        with permission._policy_lock():
+            pass
+
+        mock_filelock.assert_called_once_with(
+            permission.POLICY_UPDATE_LOCK_PATH,
+            permission.POLICY_UPDATE_LOCK_TIMEOUT_SECONDS)
+        mock_lock.__enter__.assert_called_once()
+        mock_lock.__exit__.assert_called_once()
+
+    @mock.patch('sky.users.permission.filelock.FileLock')
+    def test_policy_lock_timeout_exception(self, mock_filelock):
+        """Test policy lock timeout exception handling."""
+        from filelock import Timeout
+
+        mock_lock = mock.Mock()
+        mock_lock.__enter__ = mock.Mock(side_effect=Timeout('test_lock'))
+        mock_lock.__exit__ = mock.Mock(return_value=None)
+        mock_filelock.return_value = mock_lock
+
+        with pytest.raises(RuntimeError) as exc_info:
+            with permission._policy_lock():
+                pass
+
+        assert 'Failed to reload policy due to a timeout' in str(exc_info.value)
+        assert 'policy_update.lock' in str(exc_info.value)
+
+
+@pytest.mark.usefixtures("reset_permission_singleton", "cleanup_env_vars")
+class TestPermissionServiceMultiProcess:
+    """Test permission service behavior in multi-process scenarios."""
+
+    @mock.patch('sky.users.permission._policy_lock')
+    @mock.patch('sky.users.permission.sqlalchemy_adapter.Adapter')
+    @mock.patch('sky.users.permission.casbin.Enforcer')
+    @mock.patch('sky.global_user_state.SQLALCHEMY_ENGINE')
+    def test_singleton_behavior_same_process(self, mock_engine,
+                                             mock_enforcer_class,
+                                             mock_adapter_class,
+                                             mock_policy_lock):
+        """Test that PermissionService behaves as a singleton within the same process."""
+        mock_enforcer = mock.Mock()
+        mock_enforcer_class.return_value = mock_enforcer
+        mock_adapter = mock.Mock()
+        mock_adapter_class.return_value = mock_adapter
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        with mock.patch.object(permission.PermissionService,
+                               '_maybe_initialize_policies'):
+            # Create first instance
+            service1 = permission.PermissionService()
+
+            # Create second instance
+            service2 = permission.PermissionService()
+
+            # Both should share the same enforcer
+            assert service1.enforcer is service2.enforcer
+            assert service1.enforcer is mock_enforcer
+
+            # Enforcer should only be created once
+            mock_enforcer_class.assert_called_once()
+
+    @mock.patch('sky.users.permission._policy_lock')
+    @mock.patch('sky.users.rbac.get_workspace_policy_permissions')
+    @mock.patch('sky.users.rbac.get_role_permissions')
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_concurrent_initialization_no_duplicate_policies(
+            self, mock_get_users, mock_get_role_perms, mock_get_workspace_perms,
+            mock_policy_lock, mock_users):
+        """Test that concurrent initialization doesn't create duplicate policies."""
+        # This simulates multiple threads trying to initialize policies simultaneously
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        # Track calls to add_policy and add_grouping_policy
+        policy_calls = []
+        grouping_policy_calls = []
+
+        def track_add_policy(*args):
+            policy_calls.append(args)
+            return True
+
+        def track_add_grouping_policy(*args):
+            grouping_policy_calls.append(args)
+            return True
+
+        # Mock enforcer
+        mock_enforcer = mock.Mock()
+        mock_enforcer.get_policy.return_value = [
+        ]  # No existing policies initially
+        mock_enforcer.add_policy.side_effect = track_add_policy
+        mock_enforcer.add_grouping_policy.side_effect = track_add_grouping_policy
+        mock_enforcer.get_roles_for_user.return_value = []  # No existing roles
+
+        # Mock role permissions
+        mock_get_role_perms.return_value = {
+            'user': {
+                'permissions': {
+                    'blocklist': [{
+                        'path': '/workspaces/config',
+                        'method': 'POST'
+                    }]
+                }
+            }
+        }
+
+        # Mock workspace permissions
+        mock_get_workspace_perms.return_value = {
+            'workspace1': ['user1', 'user2']
+        }
+
+        # Mock users
+        mock_get_users.return_value = mock_users
+
+        # Simulate concurrent initialization
+        services = []
+        errors = []
+
+        def create_service():
+            try:
+                with mock.patch.object(permission.PermissionService,
+                                       '__init__',
+                                       return_value=None):
+                    service = permission.PermissionService()
+                    service.enforcer = mock_enforcer
+                    service._maybe_initialize_policies()
+                    services.append(service)
+            except Exception as e:
+                errors.append(e)
+
+        # Create multiple threads to simulate concurrent access
+        threads = []
+        for _ in range(5):
+            thread = threading.Thread(target=create_service)
+            threads.append(thread)
+
+        # Start all threads simultaneously
+        for thread in threads:
+            thread.start()
+
+        # Wait for all threads to complete
+        for thread in threads:
+            thread.join()
+
+        # Verify no errors occurred
+        assert len(
+            errors) == 0, f"Errors during concurrent initialization: {errors}"
+
+        # Verify that each policy and grouping policy was only added once per user
+        # (even though multiple threads tried to initialize)
+        unique_policy_calls = set(policy_calls)
+        unique_grouping_policy_calls = set(grouping_policy_calls)
+
+        # Should have one policy call per workspace-user combination
+        expected_policy_calls = {('user', '/workspaces/config', 'POST'),
+                                 ('user1', 'workspace1', '*'),
+                                 ('user2', 'workspace1', '*')}
+
+        # Should have one grouping policy call per user (for default role assignment)
+        expected_grouping_policy_calls = {('user1', rbac.get_default_role()),
+                                          ('user2', rbac.get_default_role()),
+                                          ('user3', rbac.get_default_role())}
+
+        assert unique_policy_calls == expected_policy_calls
+        assert unique_grouping_policy_calls == expected_grouping_policy_calls
+
+    @mock.patch('sky.users.permission._policy_lock')
+    @mock.patch('sky.users.rbac.get_workspace_policy_permissions')
+    @mock.patch('sky.users.rbac.get_role_permissions')
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_idempotent_user_addition(self, mock_get_users, mock_get_role_perms,
+                                      mock_get_workspace_perms,
+                                      mock_policy_lock, mock_users):
+        """Test that adding the same user multiple times is idempotent."""
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        # Track calls to add_grouping_policy
+        grouping_policy_calls = []
+
+        def track_add_grouping_policy(*args):
+            grouping_policy_calls.append(args)
+            return True
+
+        # Mock enforcer - simulate that users already have roles after first call
+        mock_enforcer = mock.Mock()
+        mock_enforcer.get_policy.return_value = []
+        mock_enforcer.add_policy.return_value = True
+        mock_enforcer.add_grouping_policy.side_effect = track_add_grouping_policy
+
+        # First call: no roles exist
+        # Subsequent calls: roles exist
+        call_count = 0
+
+        def get_roles_side_effect(user_id):
+            nonlocal call_count
+            call_count += 1
+            if call_count <= len(mock_users):  # First time for each user
+                return []  # No roles
+            else:
+                return ['user']  # Has roles
+
+        mock_enforcer.get_roles_for_user.side_effect = get_roles_side_effect
+
+        # Mock other dependencies
+        mock_get_role_perms.return_value = {
+            'user': {
+                'permissions': {
+                    'blocklist': []
+                }
+            }
+        }
+        mock_get_workspace_perms.return_value = {}
+        mock_get_users.return_value = mock_users
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+
+            # Initialize policies multiple times
+            service._maybe_initialize_policies()
+            service._maybe_initialize_policies()
+            service._maybe_initialize_policies()
+
+            # Each user should only be added once (3 users total)
+            assert len(grouping_policy_calls) == len(mock_users)
+
+            # Verify each user was added exactly once
+            expected_calls = {
+                (user.id, rbac.get_default_role()) for user in mock_users
+            }
+            actual_calls = set(grouping_policy_calls)
+            assert actual_calls == expected_calls
+
+    @mock.patch('sky.users.permission._policy_lock')
+    def test_concurrent_policy_updates_use_lock(self, mock_policy_lock):
+        """Test that concurrent policy updates properly use locks."""
+        lock_calls = []
+
+        # Create a context manager that tracks calls
+        class MockContextManager:
+
+            def __enter__(self):
+                lock_calls.append('enter')
+                return mock.Mock()
+
+            def __exit__(self, *args):
+                lock_calls.append('exit')
+                return None
+
+        mock_policy_lock.return_value = MockContextManager()
+
+        # Mock enforcer
+        mock_enforcer = mock.Mock()
+        mock_enforcer.add_policy.return_value = True
+        mock_enforcer.save_policy.return_value = True
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+
+            # Test a single workspace policy update to verify lock is used
+            service.add_workspace_policy('test-workspace', ['user1'])
+
+            # Verify that lock was acquired and released
+            assert 'enter' in lock_calls
+            assert 'exit' in lock_calls
+            assert len([call for call in lock_calls if call == 'enter']) == 1
+            assert len([call for call in lock_calls if call == 'exit']) == 1
+
+    @mock.patch('sky.users.permission._policy_lock')
+    @mock.patch('sky.users.rbac.get_workspace_policy_permissions')
+    @mock.patch('sky.users.rbac.get_role_permissions')
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_no_duplicate_policies_when_policies_already_exist(
+            self, mock_get_users, mock_get_role_perms, mock_get_workspace_perms,
+            mock_policy_lock, mock_users):
+        """Test that no duplicate policies are created when they already exist."""
+        mock_policy_lock.return_value.__enter__ = mock.Mock()
+        mock_policy_lock.return_value.__exit__ = mock.Mock()
+
+        # Mock existing policies that match what would be created
+        existing_policies = [['user', '/workspaces/config', 'POST'],
+                             ['user1', 'workspace1', '*'],
+                             ['user2', 'workspace1', '*']]
+
+        mock_enforcer = mock.Mock()
+        mock_enforcer.get_policy.return_value = existing_policies
+        mock_enforcer.add_policy.return_value = True
+        mock_enforcer.add_grouping_policy.return_value = True
+        mock_enforcer.get_roles_for_user.return_value = [
+            'user'
+        ]  # Users already have roles
+
+        # Mock dependencies to match existing policies
+        mock_get_role_perms.return_value = {
+            'user': {
+                'permissions': {
+                    'blocklist': [{
+                        'path': '/workspaces/config',
+                        'method': 'POST'
+                    }]
+                }
+            }
+        }
+
+        mock_get_workspace_perms.return_value = {
+            'workspace1': ['user1', 'user2']
+        }
+
+        mock_get_users.return_value = mock_users
+
+        with mock.patch.object(permission.PermissionService,
+                               '__init__',
+                               return_value=None):
+            service = permission.PermissionService()
+            service.enforcer = mock_enforcer
+
+            # Initialize policies when they already exist
+            service._maybe_initialize_policies()
+
+            # Since policies already exist and users already have roles,
+            # no new policies or grouping policies should be added
+            mock_enforcer.remove_filtered_policy.assert_not_called()
+
+            # The save_policy might not be called if no changes were made
+            # We mainly want to ensure no duplicate additions occurred
diff --git a/tests/unit_tests/test_sky/users/test_server.py b/tests/unit_tests/test_sky/users/test_server.py
new file mode 100644
index 000000000..7416c3f8f
--- /dev/null
+++ b/tests/unit_tests/test_sky/users/test_server.py
@@ -0,0 +1,308 @@
+"""Unit tests for the users server endpoints."""
+
+from unittest import mock
+
+import fastapi
+import pytest
+
+from sky import models
+from sky.server.requests import payloads
+from sky.skylet import constants
+from sky.users import rbac
+from sky.users import server
+from sky.utils import common
+
+
+@pytest.fixture
+def mock_users():
+    """Create mock users for testing."""
+    user1 = models.User(id='user1', name='Alice')
+    user2 = models.User(id='user2', name='Bob')
+    user3 = models.User(id='user3', name='Charlie')
+    return [user1, user2, user3]
+
+
+@pytest.fixture
+def mock_request():
+    """Create a mock FastAPI request."""
+    request = mock.MagicMock(spec=fastapi.Request)
+    request.state = mock.MagicMock()
+    return request
+
+
+class TestUsersEndpoints:
+    """Test class for users server endpoints."""
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    @mock.patch('sky.users.permission.permission_service.get_user_roles')
+    @pytest.mark.asyncio
+    async def test_users_endpoint_success(self, mock_get_user_roles,
+                                          mock_get_all_users, mock_users):
+        """Test successful GET /users endpoint."""
+        # Setup
+        mock_get_all_users.return_value = mock_users
+        mock_get_user_roles.side_effect = [
+            ['admin'],  # Alice has admin role
+            ['user'],  # Bob has user role
+            []  # Charlie has no roles
+        ]
+
+        # Execute
+        result = await server.users()
+
+        # Verify
+        assert len(result) == 3
+        assert result[0] == {'id': 'user1', 'name': 'Alice', 'role': 'admin'}
+        assert result[1] == {'id': 'user2', 'name': 'Bob', 'role': 'user'}
+        assert result[2] == {'id': 'user3', 'name': 'Charlie', 'role': ''}
+
+        # Verify function calls
+        mock_get_all_users.assert_called_once()
+        assert mock_get_user_roles.call_count == 3
+        mock_get_user_roles.assert_any_call('user1')
+        mock_get_user_roles.assert_any_call('user2')
+        mock_get_user_roles.assert_any_call('user3')
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    @mock.patch('sky.users.permission.permission_service.get_user_roles')
+    @pytest.mark.asyncio
+    async def test_users_endpoint_empty(self, mock_get_user_roles,
+                                        mock_get_all_users):
+        """Test GET /users endpoint with no users."""
+        # Setup
+        mock_get_all_users.return_value = []
+
+        # Execute
+        result = await server.users()
+
+        # Verify
+        assert result == []
+        mock_get_all_users.assert_called_once()
+        mock_get_user_roles.assert_not_called()
+
+    @mock.patch('sky.users.permission.permission_service.get_user_roles')
+    @pytest.mark.asyncio
+    async def test_get_current_user_role_authenticated_user(
+            self, mock_get_user_roles, mock_request):
+        """Test GET /users/role endpoint with authenticated user."""
+        # Setup
+        test_user = models.User(id='test_user', name='Test User')
+        mock_request.state.auth_user = test_user
+        mock_get_user_roles.return_value = ['admin']
+
+        # Execute
+        result = await server.get_current_user_role(mock_request)
+
+        # Verify
+        assert result == {'name': 'Test User', 'role': 'admin'}
+        mock_get_user_roles.assert_called_once_with('test_user')
+
+    @mock.patch('sky.users.permission.permission_service.get_user_roles')
+    @pytest.mark.asyncio
+    async def test_get_current_user_role_authenticated_user_no_roles(
+            self, mock_get_user_roles, mock_request):
+        """Test GET /users/role endpoint with authenticated user who has no roles."""
+        # Setup
+        test_user = models.User(id='test_user', name='Test User')
+        mock_request.state.auth_user = test_user
+        mock_get_user_roles.return_value = []
+
+        # Execute
+        result = await server.get_current_user_role(mock_request)
+
+        # Verify
+        assert result == {'name': 'Test User', 'role': ''}
+        mock_get_user_roles.assert_called_once_with('test_user')
+
+    @pytest.mark.asyncio
+    async def test_get_current_user_role_no_auth_user(self, mock_request):
+        """Test GET /users/role endpoint with no authenticated user."""
+        # Setup
+        mock_request.state.auth_user = None
+
+        # Execute
+        result = await server.get_current_user_role(mock_request)
+
+        # Verify - should return admin role when no auth user
+        assert result == {'name': '', 'role': rbac.RoleName.ADMIN.value}
+
+    @mock.patch('sky.users.rbac.get_supported_roles')
+    @mock.patch('sky.global_user_state.get_user')
+    @mock.patch('sky.users.permission.permission_service.update_role')
+    @pytest.mark.asyncio
+    async def test_user_update_success(self, mock_update_role, mock_get_user,
+                                       mock_get_supported_roles):
+        """Test successful POST /users/update endpoint."""
+        # Setup
+        mock_get_supported_roles.return_value = ['admin', 'user']
+        test_user = models.User(id='test_user', name='Test User')
+        mock_get_user.return_value = test_user
+
+        update_body = payloads.UserUpdateBody(user_id='test_user', role='admin')
+
+        # Execute
+        result = await server.user_update(update_body)
+
+        # Verify
+        assert result is None  # Function returns None on success
+        mock_get_supported_roles.assert_called_once()
+        mock_get_user.assert_called_once_with('test_user')
+        mock_update_role.assert_called_once_with('test_user', 'admin')
+
+    @mock.patch('sky.users.rbac.get_supported_roles')
+    @pytest.mark.asyncio
+    async def test_user_update_invalid_role(self, mock_get_supported_roles):
+        """Test POST /users/update endpoint with invalid role."""
+        # Setup
+        mock_get_supported_roles.return_value = ['admin', 'user']
+
+        update_body = payloads.UserUpdateBody(user_id='test_user',
+                                              role='invalid_role')
+
+        # Execute & Verify
+        with pytest.raises(fastapi.HTTPException) as exc_info:
+            await server.user_update(update_body)
+
+        assert exc_info.value.status_code == 400
+        assert 'Invalid role: invalid_role' in str(exc_info.value.detail)
+        mock_get_supported_roles.assert_called_once()
+
+    @mock.patch('sky.users.rbac.get_supported_roles')
+    @mock.patch('sky.global_user_state.get_user')
+    @pytest.mark.asyncio
+    async def test_user_update_user_not_found(self, mock_get_user,
+                                              mock_get_supported_roles):
+        """Test POST /users/update endpoint with non-existent user."""
+        # Setup
+        mock_get_supported_roles.return_value = ['admin', 'user']
+        mock_get_user.return_value = None
+
+        update_body = payloads.UserUpdateBody(user_id='nonexistent_user',
+                                              role='admin')
+
+        # Execute & Verify
+        with pytest.raises(fastapi.HTTPException) as exc_info:
+            await server.user_update(update_body)
+
+        assert exc_info.value.status_code == 400
+        assert 'User nonexistent_user does not exist' in str(
+            exc_info.value.detail)
+        mock_get_supported_roles.assert_called_once()
+        mock_get_user.assert_called_once_with('nonexistent_user')
+
+    @mock.patch('sky.users.rbac.get_supported_roles')
+    @mock.patch('sky.global_user_state.get_user')
+    @pytest.mark.asyncio
+    async def test_user_update_server_user_forbidden(self, mock_get_user,
+                                                     mock_get_supported_roles):
+        """Test POST /users/update endpoint with server internal user."""
+        # Setup
+        mock_get_supported_roles.return_value = ['admin', 'user']
+        server_user = models.User(id=common.SERVER_ID, name='Server User')
+        mock_get_user.return_value = server_user
+
+        update_body = payloads.UserUpdateBody(user_id=common.SERVER_ID,
+                                              role='user')
+
+        # Execute & Verify
+        with pytest.raises(fastapi.HTTPException) as exc_info:
+            await server.user_update(update_body)
+
+        assert exc_info.value.status_code == 400
+        assert 'Cannot update role for internal API server user' in str(
+            exc_info.value.detail)
+        mock_get_supported_roles.assert_called_once()
+        mock_get_user.assert_called_once_with(common.SERVER_ID)
+
+    @mock.patch('sky.users.rbac.get_supported_roles')
+    @mock.patch('sky.global_user_state.get_user')
+    @pytest.mark.asyncio
+    async def test_user_update_system_user_forbidden(self, mock_get_user,
+                                                     mock_get_supported_roles):
+        """Test POST /users/update endpoint with system user."""
+        # Setup
+        mock_get_supported_roles.return_value = ['admin', 'user']
+        system_user = models.User(id=constants.SKYPILOT_SYSTEM_USER_ID,
+                                  name='System User')
+        mock_get_user.return_value = system_user
+
+        update_body = payloads.UserUpdateBody(
+            user_id=constants.SKYPILOT_SYSTEM_USER_ID, role='user')
+
+        # Execute & Verify
+        with pytest.raises(fastapi.HTTPException) as exc_info:
+            await server.user_update(update_body)
+
+        assert exc_info.value.status_code == 400
+        assert 'Cannot update role for internal API server user' in str(
+            exc_info.value.detail)
+        mock_get_supported_roles.assert_called_once()
+        mock_get_user.assert_called_once_with(constants.SKYPILOT_SYSTEM_USER_ID)
+
+    @mock.patch('sky.users.rbac.get_supported_roles')
+    @mock.patch('sky.global_user_state.get_user')
+    @mock.patch('sky.users.permission.permission_service.update_role')
+    @pytest.mark.asyncio
+    async def test_user_update_update_role_exception(self, mock_update_role,
+                                                     mock_get_user,
+                                                     mock_get_supported_roles):
+        """Test POST /users/update endpoint when update_role raises exception."""
+        # Setup
+        mock_get_supported_roles.return_value = ['admin', 'user']
+        test_user = models.User(id='test_user', name='Test User')
+        mock_get_user.return_value = test_user
+        mock_update_role.side_effect = Exception('Database error')
+
+        update_body = payloads.UserUpdateBody(user_id='test_user', role='admin')
+
+        # Execute & Verify
+        with pytest.raises(Exception) as exc_info:
+            await server.user_update(update_body)
+
+        assert 'Database error' in str(exc_info.value)
+        mock_get_supported_roles.assert_called_once()
+        mock_get_user.assert_called_once_with('test_user')
+        mock_update_role.assert_called_once_with('test_user', 'admin')
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    @mock.patch('sky.users.permission.permission_service.get_user_roles')
+    @pytest.mark.asyncio
+    async def test_users_endpoint_with_multiple_roles(self, mock_get_user_roles,
+                                                      mock_get_all_users,
+                                                      mock_users):
+        """Test GET /users endpoint when user has multiple roles."""
+        # Setup
+        mock_get_all_users.return_value = mock_users
+        mock_get_user_roles.side_effect = [
+            ['admin', 'user'],  # Alice has multiple roles - should return first
+            ['user'],  # Bob has single role
+            ['viewer',
+             'guest']  # Charlie has multiple roles - should return first
+        ]
+
+        # Execute
+        result = await server.users()
+
+        # Verify - should return the first role in the list
+        assert len(result) == 3
+        assert result[0] == {'id': 'user1', 'name': 'Alice', 'role': 'admin'}
+        assert result[1] == {'id': 'user2', 'name': 'Bob', 'role': 'user'}
+        assert result[2] == {'id': 'user3', 'name': 'Charlie', 'role': 'viewer'}
+
+    @mock.patch('sky.users.permission.permission_service.get_user_roles')
+    @pytest.mark.asyncio
+    async def test_get_current_user_role_multiple_roles(self,
+                                                        mock_get_user_roles,
+                                                        mock_request):
+        """Test GET /users/role endpoint when user has multiple roles."""
+        # Setup
+        test_user = models.User(id='test_user', name='Test User')
+        mock_request.state.auth_user = test_user
+        mock_get_user_roles.return_value = ['admin', 'user', 'viewer']
+
+        # Execute
+        result = await server.get_current_user_role(mock_request)
+
+        # Verify - should return the first role in the list
+        assert result == {'name': 'Test User', 'role': 'admin'}
+        mock_get_user_roles.assert_called_once_with('test_user')
diff --git a/tests/unit_tests/test_sky/workspaces/test_workspace_permissions.py b/tests/unit_tests/test_sky/workspaces/test_workspace_permissions.py
new file mode 100644
index 000000000..e3c1fbb04
--- /dev/null
+++ b/tests/unit_tests/test_sky/workspaces/test_workspace_permissions.py
@@ -0,0 +1,210 @@
+"""Unit tests for workspace permissions."""
+
+import unittest
+from unittest import mock
+
+from sky import models
+from sky.workspaces import utils as workspaces_utils
+
+
+class TestWorkspacePermissions(unittest.TestCase):
+    """Test workspace permission functionality."""
+
+    def setUp(self):
+        """Set up test environment."""
+        # Create mock users
+        self.user1 = models.User(id='user1', name='Alice')
+        self.user2 = models.User(id='user2', name='Bob')
+        self.user3 = models.User(id='user3', name='Charlie')
+        # Create users with duplicate names to test conflict resolution
+        self.user4 = models.User(id='user4', name='Alice')  # Same name as user1
+        self.user5 = models.User(id='user5', name=None)  # User with no name
+        self.all_users = [
+            self.user1, self.user2, self.user3, self.user4, self.user5
+        ]
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_public_workspace_config(self, mock_get_users):
+        """Test that public workspace config returns wildcard."""
+        mock_get_users.return_value = self.all_users
+        public_config = {'private': False}
+        users = workspaces_utils.get_workspace_users(public_config)
+        self.assertEqual(users, ['*'],
+                         "Public workspace should return ['*'] for all users")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_config(self, mock_get_users):
+        """Test that private workspace config returns specific users."""
+        mock_get_users.return_value = self.all_users
+        private_config = {'private': True, 'allowed_users': ['Bob', 'Charlie']}
+        users = workspaces_utils.get_workspace_users(private_config)
+        expected_users = ['user2', 'user3']  # user IDs for Bob and Charlie
+        self.assertEqual(set(users), set(expected_users),
+                         "Private workspace should return specific user IDs")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_with_unknown_user(self, mock_get_users):
+        """Test private workspace with unknown user."""
+        mock_get_users.return_value = self.all_users
+        private_config = {
+            'private': True,
+            'allowed_users': ['Bob', 'UnknownUser']
+        }
+        users = workspaces_utils.get_workspace_users(private_config)
+        expected_users = ['user2']  # Only Bob's user ID
+        self.assertEqual(
+            users, expected_users,
+            "Unknown users should be ignored in private workspace")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_workspace_default_public(self, mock_get_users):
+        """Test that workspace without 'private' key defaults to public."""
+        mock_get_users.return_value = self.all_users
+        default_config = {}
+        users = workspaces_utils.get_workspace_users(default_config)
+        self.assertEqual(
+            users, ['*'],
+            "Workspace without 'private' key should default to public")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_with_user_ids(self, mock_get_users):
+        """Test private workspace using user IDs directly."""
+        mock_get_users.return_value = self.all_users
+        private_config = {'private': True, 'allowed_users': ['user1', 'user2']}
+        users = workspaces_utils.get_workspace_users(private_config)
+        expected_users = ['user1', 'user2']
+        self.assertEqual(set(users), set(expected_users),
+                         "Private workspace should accept user IDs directly")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_mixed_names_and_ids(self, mock_get_users):
+        """Test private workspace with mix of user names and IDs."""
+        mock_get_users.return_value = self.all_users
+        private_config = {
+            'private': True,
+            'allowed_users': ['user1', 'Bob', 'user3']
+        }
+        users = workspaces_utils.get_workspace_users(private_config)
+        expected_users = ['user1', 'user2', 'user3']
+        self.assertEqual(
+            set(users), set(expected_users),
+            "Private workspace should handle mix of names and IDs")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_duplicate_user_names_raises_error(
+            self, mock_get_users):
+        """Test private workspace with duplicate user names raises ValueError."""
+        mock_get_users.return_value = self.all_users
+        private_config = {'private': True, 'allowed_users': ['Alice']}
+
+        with self.assertRaises(ValueError) as context:
+            workspaces_utils.get_workspace_users(private_config)
+
+        self.assertIn('User \'Alice\' has multiple IDs', str(context.exception))
+        self.assertIn('user1, user4', str(context.exception))
+        self.assertIn('Please specify the user ID instead',
+                      str(context.exception))
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_empty_allowed_users(self, mock_get_users):
+        """Test private workspace with empty allowed_users list."""
+        mock_get_users.return_value = self.all_users
+        private_config = {'private': True, 'allowed_users': []}
+        users = workspaces_utils.get_workspace_users(private_config)
+        self.assertEqual(
+            users, [],
+            "Private workspace with empty allowed_users should return empty list"
+        )
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_no_allowed_users_key(self, mock_get_users):
+        """Test private workspace without allowed_users key."""
+        mock_get_users.return_value = self.all_users
+        private_config = {'private': True}
+        users = workspaces_utils.get_workspace_users(private_config)
+        self.assertEqual(
+            users, [],
+            "Private workspace without allowed_users should return empty list")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_only_unknown_users(self, mock_get_users):
+        """Test private workspace with only unknown users."""
+        mock_get_users.return_value = self.all_users
+        private_config = {
+            'private': True,
+            'allowed_users': ['UnknownUser1', 'UnknownUser2']
+        }
+        users = workspaces_utils.get_workspace_users(private_config)
+        self.assertEqual(
+            users, [],
+            "Private workspace with only unknown users should return empty list"
+        )
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_user_with_no_name(self, mock_get_users):
+        """Test private workspace with user that has no name."""
+        mock_get_users.return_value = self.all_users
+        # Try to use the ID of a user with no name
+        private_config = {'private': True, 'allowed_users': ['user5']}
+        users = workspaces_utils.get_workspace_users(private_config)
+        self.assertEqual(
+            users, ['user5'],
+            "Private workspace should accept user ID even if user has no name")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_unique_user_name(self, mock_get_users):
+        """Test private workspace with unique user name (no duplicates)."""
+        mock_get_users.return_value = [self.user2,
+                                       self.user3]  # Only Bob and Charlie
+        private_config = {'private': True, 'allowed_users': ['Bob']}
+        users = workspaces_utils.get_workspace_users(private_config)
+        self.assertEqual(
+            users, ['user2'],
+            "Private workspace should resolve unique user name to ID")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    @mock.patch('sky.workspaces.utils.logger')
+    def test_private_workspace_logs_warning_for_unknown_user(
+            self, mock_logger, mock_get_users):
+        """Test that warning is logged for unknown users."""
+        mock_get_users.return_value = self.all_users
+        private_config = {
+            'private': True,
+            'allowed_users': ['Bob', 'UnknownUser']
+        }
+
+        users = workspaces_utils.get_workspace_users(private_config)
+
+        # Should return Bob's ID
+        self.assertEqual(users, ['user2'])
+
+        # Should log warning for unknown user
+        mock_logger.warning.assert_called_once_with(
+            "User 'UnknownUser' not found in all users")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_private_workspace_preserves_order(self, mock_get_users):
+        """Test that private workspace preserves order of allowed users."""
+        mock_get_users.return_value = self.all_users
+        private_config = {
+            'private': True,
+            'allowed_users': ['Charlie', 'Bob', 'user1']
+        }
+        users = workspaces_utils.get_workspace_users(private_config)
+        expected_users = ['user3', 'user2', 'user1']  # Charlie, Bob, user1
+        self.assertEqual(
+            users, expected_users,
+            "Private workspace should preserve order of allowed users")
+
+    @mock.patch('sky.global_user_state.get_all_users')
+    def test_empty_all_users_list(self, mock_get_users):
+        """Test behavior when get_all_users returns empty list."""
+        mock_get_users.return_value = []
+        private_config = {'private': True, 'allowed_users': ['Alice', 'Bob']}
+        users = workspaces_utils.get_workspace_users(private_config)
+        self.assertEqual(
+            users, [], "Should return empty list when no users exist in system")
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/unit_tests/test_sky/workspaces/test_workspace_user_permissions.py b/tests/unit_tests/test_sky/workspaces/test_workspace_user_permissions.py
new file mode 100644
index 000000000..a2b3dc628
--- /dev/null
+++ b/tests/unit_tests/test_sky/workspaces/test_workspace_user_permissions.py
@@ -0,0 +1,433 @@
+"""Integration tests for workspace permissions with user roles.
+
+This module tests that users with "user" role can only see resources
+(clusters and managed jobs) in workspaces they have permission to access.
+"""
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+
+import pytest
+
+from sky import models
+from sky.jobs.server import core as managed_jobs_core
+from sky.skylet import constants
+from sky.users import permission
+from sky.utils import common_utils
+from sky.workspaces import core as workspaces_core
+
+
+class TestWorkspaceUserPermissions(unittest.TestCase):
+    """Test workspace permissions for users with different roles."""
+
+    def setUp(self):
+        """Set up test environment."""
+        # Mock users
+        self.admin_user = models.User(id='admin_user_id', name='admin')
+        self.user1 = models.User(id='user1_id', name='user1')
+        self.user2 = models.User(id='user2_id', name='user2')
+        self.user3 = models.User(id='user3_id', name='user3')
+
+        # Mock clusters in different workspaces
+        self.clusters = [{
+            'name': 'public-cluster-1',
+            'workspace': 'public-ws',
+            'status': 'UP',
+            'user_hash': 'user1_id'
+        }, {
+            'name': 'public-cluster-2',
+            'workspace': 'public-ws',
+            'status': 'UP',
+            'user_hash': 'user2_id'
+        }, {
+            'name': 'private-cluster-1',
+            'workspace': 'private-ws-1',
+            'status': 'UP',
+            'user_hash': 'user1_id'
+        }, {
+            'name': 'private-cluster-2',
+            'workspace': 'private-ws-2',
+            'status': 'UP',
+            'user_hash': 'user2_id'
+        }, {
+            'name': 'default-cluster',
+            'workspace': constants.SKYPILOT_DEFAULT_WORKSPACE,
+            'status': 'UP',
+            'user_hash': 'user3_id'
+        }]
+
+        # Mock managed jobs in different workspaces
+        self.managed_jobs = [{
+            'job_id': 1,
+            'job_name': 'public-job-1',
+            'workspace': 'public-ws',
+            'user_hash': 'user1_id',
+            'status': mock.MagicMock(is_terminal=lambda: False)
+        }, {
+            'job_id': 2,
+            'job_name': 'public-job-2',
+            'workspace': 'public-ws',
+            'user_hash': 'user2_id',
+            'status': mock.MagicMock(is_terminal=lambda: False)
+        }, {
+            'job_id': 3,
+            'job_name': 'private-job-1',
+            'workspace': 'private-ws-1',
+            'user_hash': 'user1_id',
+            'status': mock.MagicMock(is_terminal=lambda: False)
+        }, {
+            'job_id': 4,
+            'job_name': 'private-job-2',
+            'workspace': 'private-ws-2',
+            'user_hash': 'user2_id',
+            'status': mock.MagicMock(is_terminal=lambda: False)
+        }, {
+            'job_id': 5,
+            'job_name': 'default-job',
+            'workspace': constants.SKYPILOT_DEFAULT_WORKSPACE,
+            'user_hash': 'user3_id',
+            'status': mock.MagicMock(is_terminal=lambda: False)
+        }]
+
+    @mock.patch.dict(os.environ, {constants.ENV_VAR_IS_SKYPILOT_SERVER: 'true'})
+    @mock.patch('sky.global_user_state.get_clusters')
+    @mock.patch('sky.workspaces.core.get_workspaces')
+    @mock.patch('sky.users.permission.permission_service')
+    def test_user_can_only_see_accessible_workspace_clusters(
+            self, mock_permission_service, mock_get_workspaces,
+            mock_get_clusters):
+        """Test that user role can only see clusters in accessible workspaces."""
+        # Mock clusters data
+        mock_get_clusters.return_value = self.clusters
+
+        # Mock permission service for user1 (user role)
+        mock_permission_service.check_workspace_permission.side_effect = (
+            lambda user_id, workspace: (
+                # user1 has access to public-ws, private-ws-1, and default
+                workspace in [
+                    'public-ws', 'private-ws-1', constants.
+                    SKYPILOT_DEFAULT_WORKSPACE
+                ] if user_id == 'user1_id' else
+                # user2 has access to public-ws, private-ws-2, and default
+                workspace in [
+                    'public-ws', 'private-ws-2', constants.
+                    SKYPILOT_DEFAULT_WORKSPACE
+                ] if user_id == 'user2_id' else
+                # user3 has access to public-ws and default only
+                workspace in
+                ['public-ws', constants.SKYPILOT_DEFAULT_WORKSPACE]
+                if user_id == 'user3_id' else False))
+
+        # Mock get_workspaces to return only accessible workspaces for user1
+        mock_get_workspaces.return_value = {
+            'public-ws': {},  # Public workspace
+            'private-ws-1': {
+                'private': True,
+                'allowed_users': ['user1_id']
+            },
+            constants.SKYPILOT_DEFAULT_WORKSPACE: {}  # Default is public
+            # Note: private-ws-2 should NOT be in this list for user1
+        }
+
+        # Test user1 can see clusters in accessible workspaces
+        with mock.patch('sky.utils.common_utils.get_current_user',
+                        return_value=self.user1):
+            from sky.backends import backend_utils
+
+            # Get accessible workspaces for user1
+            accessible_ws = workspaces_core.get_workspaces()
+
+            # Filter clusters by accessible workspaces
+            filtered_clusters = [
+                c for c in self.clusters
+                if c.get('workspace', constants.SKYPILOT_DEFAULT_WORKSPACE) in
+                accessible_ws
+            ]
+
+            # user1 should see: public-cluster-1, public-cluster-2,
+            # private-cluster-1, default-cluster
+            expected_cluster_names = {
+                'public-cluster-1', 'public-cluster-2', 'private-cluster-1',
+                'default-cluster'
+            }
+            actual_cluster_names = {c['name'] for c in filtered_clusters}
+
+            self.assertEqual(
+                actual_cluster_names, expected_cluster_names,
+                "user1 should see clusters from public-ws, "
+                "private-ws-1, and default workspace only")
+
+    @mock.patch.dict(os.environ, {constants.ENV_VAR_IS_SKYPILOT_SERVER: 'true'})
+    @mock.patch('sky.workspaces.core.get_workspaces')
+    @mock.patch('sky.users.permission.permission_service')
+    def test_user_cannot_see_private_workspace_clusters(self,
+                                                        mock_permission_service,
+                                                        mock_get_workspaces):
+        """Test that user role cannot see clusters in inaccessible private workspaces."""
+        # Mock permission service for user3 (only access to public workspaces)
+        mock_permission_service.check_workspace_permission.side_effect = (
+            lambda user_id, workspace: workspace in
+            ['public-ws', constants.SKYPILOT_DEFAULT_WORKSPACE]
+            if user_id == 'user3_id' else False)
+
+        # Mock get_workspaces to return only accessible workspaces for user3
+        mock_get_workspaces.return_value = {
+            'public-ws': {},  # Public workspace
+            constants.SKYPILOT_DEFAULT_WORKSPACE: {}
+            # Note: private workspaces should NOT be in this list for user3
+        }
+
+        # Test user3 cannot see private workspace clusters
+        with mock.patch('sky.utils.common_utils.get_current_user',
+                        return_value=self.user3):
+            accessible_ws = workspaces_core.get_workspaces()
+
+            # user3 should not have access to private workspaces
+            self.assertNotIn('private-ws-1', accessible_ws,
+                             "user3 should not have access to private-ws-1")
+            self.assertNotIn('private-ws-2', accessible_ws,
+                             "user3 should not have access to private-ws-2")
+
+            # user3 should have access to public workspaces
+            self.assertIn('public-ws', accessible_ws,
+                          "user3 should have access to public-ws")
+            self.assertIn(constants.SKYPILOT_DEFAULT_WORKSPACE, accessible_ws,
+                          "user3 should have access to default workspace")
+
+    @mock.patch.dict(os.environ, {constants.ENV_VAR_IS_SKYPILOT_SERVER: 'true'})
+    @mock.patch('sky.workspaces.core.get_workspaces')
+    @mock.patch('sky.users.permission.permission_service')
+    def test_managed_jobs_workspace_filtering(self, mock_permission_service,
+                                              mock_get_workspaces):
+        """Test that managed jobs are filtered by workspace permissions."""
+        # Mock permission service
+        mock_permission_service.check_workspace_permission.side_effect = (
+            lambda user_id, workspace:
+            (workspace in [
+                'public-ws', 'private-ws-1', constants.
+                SKYPILOT_DEFAULT_WORKSPACE
+            ] if user_id == 'user1_id' else workspace in [
+                'public-ws', 'private-ws-2', constants.
+                SKYPILOT_DEFAULT_WORKSPACE
+            ] if user_id == 'user2_id' else workspace in
+             ['public-ws', constants.SKYPILOT_DEFAULT_WORKSPACE]
+             if user_id == 'user3_id' else False))
+
+        # Mock get_workspaces to return accessible workspaces for user1
+        mock_get_workspaces.return_value = {
+            'public-ws': {},
+            'private-ws-1': {
+                'private': True,
+                'allowed_users': ['user1_id']
+            },
+            constants.SKYPILOT_DEFAULT_WORKSPACE: {}
+            # Note: private-ws-2 should NOT be in this list for user1
+        }
+
+        # Test user1 can see jobs in accessible workspaces
+        with mock.patch('sky.utils.common_utils.get_current_user',
+                        return_value=self.user1):
+            accessible_ws = workspaces_core.get_workspaces()
+
+            # Filter jobs by accessible workspaces (similar to
+            # managed_jobs_core.queue)
+            filtered_jobs = [
+                job for job in self.managed_jobs
+                if job.get('workspace', constants.SKYPILOT_DEFAULT_WORKSPACE) in
+                accessible_ws
+            ]
+
+            # user1 should see: public-job-1, public-job-2,
+            # private-job-1, default-job
+            expected_job_names = {
+                'public-job-1', 'public-job-2', 'private-job-1', 'default-job'
+            }
+            actual_job_names = {job['job_name'] for job in filtered_jobs}
+
+            self.assertEqual(
+                actual_job_names, expected_job_names,
+                "user1 should see jobs from accessible "
+                "workspaces only")
+
+    @mock.patch.dict(os.environ, {constants.ENV_VAR_IS_SKYPILOT_SERVER: 'true'})
+    @mock.patch('sky.workspaces.core.get_workspaces')
+    @mock.patch('sky.users.permission.permission_service')
+    def test_user_cannot_see_private_workspace_jobs(self,
+                                                    mock_permission_service,
+                                                    mock_get_workspaces):
+        """Test that user role cannot see jobs in inaccessible private workspaces."""
+        # Mock permission service for user3
+        mock_permission_service.check_workspace_permission.side_effect = (
+            lambda user_id, workspace: workspace in
+            ['public-ws', constants.SKYPILOT_DEFAULT_WORKSPACE]
+            if user_id == 'user3_id' else False)
+
+        # Mock get_workspaces to return accessible workspaces for user3
+        mock_get_workspaces.return_value = {
+            'public-ws': {},
+            constants.SKYPILOT_DEFAULT_WORKSPACE: {}
+            # Note: private workspaces should NOT be in this list for user3
+        }
+
+        # Test user3 can only see jobs from accessible workspaces
+        with mock.patch('sky.utils.common_utils.get_current_user',
+                        return_value=self.user3):
+            accessible_ws = workspaces_core.get_workspaces()
+
+            # Filter jobs by accessible workspaces
+            filtered_jobs = [
+                job for job in self.managed_jobs
+                if job.get('workspace', constants.SKYPILOT_DEFAULT_WORKSPACE) in
+                accessible_ws
+            ]
+
+            # user3 should only see: public-job-1, public-job-2, default-job
+            # Should NOT see: private-job-1, private-job-2
+            expected_job_names = {'public-job-1', 'public-job-2', 'default-job'}
+            actual_job_names = {job['job_name'] for job in filtered_jobs}
+
+            self.assertEqual(
+                actual_job_names, expected_job_names,
+                "user3 should only see jobs from public and "
+                "default workspaces")
+
+            # Explicitly check that private jobs are not visible
+            private_job_names = {'private-job-1', 'private-job-2'}
+            self.assertTrue(private_job_names.isdisjoint(actual_job_names),
+                            "user3 should not see private workspace jobs")
+
+    @mock.patch.dict(os.environ, {constants.ENV_VAR_IS_SKYPILOT_SERVER: 'true'})
+    @mock.patch('sky.workspaces.core.get_workspaces')
+    @mock.patch('sky.users.permission.permission_service')
+    def test_admin_user_can_see_all_workspaces(self, mock_permission_service,
+                                               mock_get_workspaces):
+        """Test that admin role can see resources in all workspaces."""
+        # Mock get_workspaces to return all workspaces for admin
+        mock_get_workspaces.return_value = {
+            'public-ws': {},
+            'private-ws-1': {
+                'private': True,
+                'allowed_users': ['user1_id']
+            },
+            'private-ws-2': {
+                'private': True,
+                'allowed_users': ['user2_id']
+            },
+            constants.SKYPILOT_DEFAULT_WORKSPACE: {}
+        }
+
+        # Mock permission service - admin should have access to all workspaces
+        mock_permission_service.check_workspace_permission.return_value = True
+
+        # Test admin can see all workspace resources
+        with mock.patch('sky.utils.common_utils.get_current_user',
+                        return_value=self.admin_user):
+            accessible_ws = workspaces_core.get_workspaces()
+
+            # Admin should have access to all workspaces
+            expected_workspaces = {
+                'public-ws', 'private-ws-1', 'private-ws-2',
+                constants.SKYPILOT_DEFAULT_WORKSPACE
+            }
+            self.assertEqual(set(accessible_ws.keys()), expected_workspaces,
+                             "admin should have access to all workspaces")
+
+            # Admin should see all clusters
+            filtered_clusters = [
+                c for c in self.clusters
+                if c.get('workspace', constants.SKYPILOT_DEFAULT_WORKSPACE) in
+                accessible_ws
+            ]
+            expected_cluster_names = {c['name'] for c in self.clusters}
+            actual_cluster_names = {c['name'] for c in filtered_clusters}
+
+            self.assertEqual(actual_cluster_names, expected_cluster_names,
+                             "admin should see all clusters")
+
+            # Admin should see all jobs
+            filtered_jobs = [
+                job for job in self.managed_jobs
+                if job.get('workspace', constants.SKYPILOT_DEFAULT_WORKSPACE) in
+                accessible_ws
+            ]
+            expected_job_names = {job['job_name'] for job in self.managed_jobs}
+            actual_job_names = {job['job_name'] for job in filtered_jobs}
+
+            self.assertEqual(actual_job_names, expected_job_names,
+                             "admin should see all managed jobs")
+
+    @mock.patch.dict(os.environ, {constants.ENV_VAR_IS_SKYPILOT_SERVER: 'true'})
+    @mock.patch('sky.workspaces.core.get_workspaces')
+    @mock.patch('sky.users.permission.permission_service')
+    def test_managed_jobs_queue_integration(self, mock_permission_service,
+                                            mock_get_workspaces):
+        """Integration test for managed_jobs_core.queue with workspace filtering."""
+        # Mock get_workspaces to return accessible workspaces for user3
+        mock_get_workspaces.return_value = {
+            'public-ws': {},
+            constants.SKYPILOT_DEFAULT_WORKSPACE: {}
+            # Note: private workspaces should NOT be in this list for user3
+        }
+
+        # Mock permission service
+        mock_permission_service.check_workspace_permission.side_effect = (
+            lambda user_id, workspace: workspace in
+            ['public-ws', constants.SKYPILOT_DEFAULT_WORKSPACE]
+            if user_id == 'user3_id' else False)
+
+        # Simulate job queue data with workspace information
+        job_queue_data = [{
+            'job_id': 1,
+            'job_name': 'public-job',
+            'workspace': 'public-ws',
+            'user_hash': 'user1_id',
+            'status': mock.MagicMock(is_terminal=lambda: False)
+        }, {
+            'job_id': 2,
+            'job_name': 'private-job',
+            'workspace': 'private-ws-1',
+            'user_hash': 'user1_id',
+            'status': mock.MagicMock(is_terminal=lambda: False)
+        }, {
+            'job_id': 3,
+            'job_name': 'default-job',
+            'workspace': constants.SKYPILOT_DEFAULT_WORKSPACE,
+            'user_hash': 'user3_id',
+            'status': mock.MagicMock(is_terminal=lambda: False)
+        }]
+
+        with mock.patch('sky.jobs.utils.load_managed_job_queue',
+                        return_value=job_queue_data):
+            with mock.patch('sky.utils.common_utils.get_current_user',
+                            return_value=self.user3):
+
+                # Test the filtering logic directly (without calling managed_jobs_core.queue
+                # to avoid complex backend mocking)
+                accessible_ws = workspaces_core.get_workspaces()
+
+                # Filter jobs by accessible workspaces (simulating what queue() does)
+                filtered_jobs = [
+                    job for job in job_queue_data
+                    if job.get('workspace', constants.SKYPILOT_DEFAULT_WORKSPACE
+                              ) in accessible_ws
+                ]
+
+                # user3 should only see jobs from accessible workspaces
+                expected_job_names = {'public-job', 'default-job'}
+                actual_job_names = {job['job_name'] for job in filtered_jobs}
+
+                self.assertEqual(
+                    actual_job_names, expected_job_names,
+                    "user3 should only see jobs from accessible "
+                    "workspaces")
+
+                # Should not see private workspace job
+                self.assertNotIn('private-job',
+                                 [job['job_name'] for job in filtered_jobs],
+                                 "user3 should not see private workspace jobs")
+
+
+if __name__ == '__main__':
+    pytest.main([__file__])