Repository: skypilot-org/skypilot
Pull request #5651 (2025-05-20 21:16:06)

 add oauth2-proxy to the helm chart
<!-- Describe the changes in this PR -->
TODO: Whichever one is merged second, add a link from oauth2 proxy page to helm chart reference

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [x] Code formatting: install pre-commit (auto-check on commit) or `bash format.sh`
- [x] Manually deployed and tested helm chart.
- [x] Verified helm chart output is equivalent to master when oauth2-proxy is disabled.
<!-- CI commands (/-prefixed) can only be triggered by repo members -->



Patch:
diff --git a/.github/workflows/release-publish.yml b/.github/workflows/release-publish.yml
index 28f904936..d717731b8 100644
--- a/.github/workflows/release-publish.yml
+++ b/.github/workflows/release-publish.yml
@@ -14,7 +14,7 @@ jobs:
       version: ${{ steps.extract_version.outputs.version }}
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           ref: ${{ github.base_ref }}
 
@@ -32,7 +32,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           ref: ${{ github.base_ref }}
 
@@ -119,7 +119,7 @@ jobs:
       package_name: 'skypilot'
       version: ${{ needs.extract-version.outputs.version }}
     secrets: inherit
-  
+
   publish-helm:
     needs: [extract-version, publish-to-pypi, trigger-docker-build]
     uses: ./.github/workflows/publish-helm.yml
@@ -129,13 +129,39 @@ jobs:
       package_name: 'skypilot'
     secrets: inherit
 
+  create-release:
+    needs: [extract-version, publish-to-pypi]
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Create and push tag
+        run: |
+          VERSION="${{ needs.extract-version.outputs.version }}"
+          git config --global user.name "GitHub Action"
+          git config --global user.email "action@github.com"
+          git tag -a "v${VERSION}" -m "Release v${VERSION}"
+          git push origin "v${VERSION}"
+
+      - name: Create GitHub Release
+        uses: softprops/action-gh-release@v2
+        with:
+          name: "Release v${{ needs.extract-version.outputs.version }}"
+          tag_name: "v${{ needs.extract-version.outputs.version }}"
+          generate_release_notes: true
+          draft: true
+
+
   cleanup-branches:
     needs: extract-version
     if: github.event.pull_request.merged == false && startsWith(github.head_ref, 'test_releases/')
     runs-on: ubuntu-latest
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           fetch-depth: 0
           token: ${{ secrets.GITHUB_TOKEN }}
diff --git a/charts/skypilot/templates/auth.yaml b/charts/skypilot/templates/auth.yaml
index 70c492225..b2df2a73a 100644
--- a/charts/skypilot/templates/auth.yaml
+++ b/charts/skypilot/templates/auth.yaml
@@ -1,4 +1,4 @@
-{{- if and (not .Values.ingress.authSecret) .Values.ingress.authCredentials }}
+{{- if and (not .Values.ingress.authSecret) .Values.ingress.authCredentials (not (index .Values.ingress "oauth2-proxy" "enabled")) }}
 apiVersion: v1
 kind: Secret
 metadata:
diff --git a/charts/skypilot/templates/ingress.yaml b/charts/skypilot/templates/ingress.yaml
index 0c6cd6eeb..cdf21e7cf 100644
--- a/charts/skypilot/templates/ingress.yaml
+++ b/charts/skypilot/templates/ingress.yaml
@@ -12,13 +12,21 @@ metadata:
     {{- if not $useNewIngressClass }}
     kubernetes.io/ingress.class: nginx
     {{- end }}
+    {{- if index .Values.ingress "oauth2-proxy" "enabled" }}
+    # OAuth2 Proxy authentication
+    nginx.ingress.kubernetes.io/auth-signin: {{ if index .Values.ingress "oauth2-proxy" "use-http" }}http{{ else }}https{{ end }}://$host/oauth2/start?rd=$escaped_request_uri
+    nginx.ingress.kubernetes.io/auth-url: {{ if index .Values.ingress "oauth2-proxy" "use-http" }}http{{ else }}https{{ end }}://$host/oauth2/auth
+    {{- else }}
+    # Basic authentication
     nginx.ingress.kubernetes.io/auth-type: basic
     nginx.ingress.kubernetes.io/auth-secret: {{ .Values.ingress.authSecret | default (printf "%s-basic-auth" .Release.Name) }}
+    {{- end }}
     nginx.ingress.kubernetes.io/proxy-read-timeout: "3600"
     nginx.ingress.kubernetes.io/proxy-send-timeout: "3600"
     nginx.ingress.kubernetes.io/proxy-body-size: "1000m"
     {{- if ne .Values.ingress.path "/" }}
     nginx.ingress.kubernetes.io/rewrite-target: /$2
+    skypilot.co/ingress-path: {{ .Values.ingress.path }}
     {{- end }}
     nginx.ingress.kubernetes.io/server-snippets: |
       location / {
diff --git a/charts/skypilot/templates/oauth2-proxy-deployment.yaml b/charts/skypilot/templates/oauth2-proxy-deployment.yaml
new file mode 100644
index 000000000..4bfbab244
--- /dev/null
+++ b/charts/skypilot/templates/oauth2-proxy-deployment.yaml
@@ -0,0 +1,88 @@
+{{- if and .Values.ingress.enabled (index .Values.ingress "oauth2-proxy" "enabled") }}
+{{- $oauth2 := index .Values.ingress "oauth2-proxy" }}
+{{- if not (index $oauth2 "oidc-issuer-url") -}}
+{{- fail "ingress.oauth2-proxy.oidc-issuer-url is required when OAuth2 proxy is enabled" -}}
+{{- end -}}
+{{- if eq (index $oauth2 "client-id") "" -}}
+{{- fail "ingress.oauth2-proxy.client-id is required when OAuth2 proxy is enabled" -}}
+{{- end -}}
+{{- if eq (index $oauth2 "client-secret") "" -}}
+{{- fail "ingress.oauth2-proxy.client-secret is required when OAuth2 proxy is enabled" -}}
+{{- end -}}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  labels:
+    app: {{ .Release.Name }}-oauth2-proxy
+    skypilot.co/component: oauth2-proxy
+  name: {{ .Release.Name }}-oauth2-proxy
+  namespace: {{ .Release.Namespace }}
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: {{ .Release.Name }}-oauth2-proxy
+  template:
+    metadata:
+      labels:
+        app: {{ .Release.Name }}-oauth2-proxy
+        skypilot.co/component: oauth2-proxy
+    spec:
+      containers:
+      - name: oauth2-proxy
+        image: {{ index .Values.ingress "oauth2-proxy" "image" | default "quay.io/oauth2-proxy/oauth2-proxy:v7.9.0" }}
+        imagePullPolicy: IfNotPresent
+        args:
+        - --provider=oidc
+        - --oidc-issuer-url={{ index .Values.ingress "oauth2-proxy" "oidc-issuer-url" }}
+        - --email-domain={{ index .Values.ingress "oauth2-proxy" "email-domain" | default "*" }}
+        - --upstream=file:///dev/null
+        - --http-address=0.0.0.0:4180
+        - --skip-provider-button=true
+        - --reverse-proxy=true
+        - --session-store-type={{ index .Values.ingress "oauth2-proxy" "session-store-type" | default "redis" }}
+        {{- if not (index .Values.ingress "oauth2-proxy" "use-https" | default false) }}
+        - --cookie-secure=false
+        {{- end }}
+        {{- if (index .Values.ingress "oauth2-proxy" "cookie-refresh") }}
+        - --cookie-refresh={{ index .Values.ingress "oauth2-proxy" "cookie-refresh" }}
+        {{- end }}
+        {{- if (index .Values.ingress "oauth2-proxy" "cookie-expire") }}
+        - --cookie-expire={{ index .Values.ingress "oauth2-proxy" "cookie-expire" }}
+        {{- end }}
+        {{- if eq (index .Values.ingress "oauth2-proxy" "session-store-type" | default "redis") "redis" }}
+        {{- if (index .Values.ingress "oauth2-proxy" "redis-url") }}
+        - --redis-connection-url={{ index .Values.ingress "oauth2-proxy" "redis-url" }}
+        {{- else }}
+        - --redis-connection-url=redis://$(SERVICE_NAME)-oauth2-proxy-redis:6379/0
+        {{- end }}
+        {{- end }}
+        env:
+        - name: OAUTH2_PROXY_CLIENT_ID
+          value: {{ index .Values.ingress "oauth2-proxy" "client-id" | quote }}
+        - name: OAUTH2_PROXY_CLIENT_SECRET
+          value: {{ index .Values.ingress "oauth2-proxy" "client-secret" | quote }}
+        - name: SERVICE_NAME
+          value: {{ .Release.Name }}
+        - name: OAUTH2_PROXY_COOKIE_SECRET
+          {{- /* Look up existing deployment to get the current cookie secret if available */}}
+          {{- $deploymentName := printf "%s-oauth2-proxy" .Release.Name }}
+          {{- $existingDeployment := lookup "apps/v1" "Deployment" .Release.Namespace $deploymentName }}
+          {{- if and $existingDeployment $existingDeployment.spec.template.spec.containers }}
+            {{- range $container := $existingDeployment.spec.template.spec.containers }}
+              {{- if eq $container.name "oauth2-proxy" }}
+                {{- range $env := $container.env }}
+                  {{- if eq $env.name "OAUTH2_PROXY_COOKIE_SECRET" }}
+          value: {{ $env.value }}
+                  {{- end }}
+                {{- end }}
+              {{- end }}
+            {{- end }}
+          {{- else }}
+          # Generate a random cookie secret if no existing one is found
+          value: {{ randAlphaNum 32 | b64enc | quote }}
+          {{- end }}
+        ports:
+        - containerPort: 4180
+          protocol: TCP
+{{- end }}
diff --git a/charts/skypilot/templates/oauth2-proxy-ingress.yaml b/charts/skypilot/templates/oauth2-proxy-ingress.yaml
new file mode 100644
index 000000000..7734d1126
--- /dev/null
+++ b/charts/skypilot/templates/oauth2-proxy-ingress.yaml
@@ -0,0 +1,33 @@
+{{- if and .Values.ingress.enabled (index .Values.ingress "oauth2-proxy" "enabled") }}
+{{- $kubeVersion := .Capabilities.KubeVersion.Major -}}
+{{- $kubeMinorVersion := .Capabilities.KubeVersion.Minor | trimSuffix "+" | int -}}
+{{- $useNewIngressClass := or (gt ($kubeVersion | int) 1) (and (eq ($kubeVersion | int) 1) (ge $kubeMinorVersion 18)) -}}
+
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: {{ .Release.Name }}-oauth2-proxy
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ .Release.Name }}-oauth2-proxy
+    skypilot.co/component: oauth2-proxy
+  annotations:
+    {{- if not $useNewIngressClass }}
+    kubernetes.io/ingress.class: nginx
+    {{- end }}
+spec:
+  {{- if $useNewIngressClass }}
+  ingressClassName: nginx
+  {{- end }}
+  rules:
+  - host: {{ .Values.ingress.host }}
+    http:
+      paths:
+      - path: /oauth2
+        pathType: Prefix
+        backend:
+          service:
+            name: {{ .Release.Name }}-oauth2-proxy
+            port:
+              number: 4180
+{{- end }}
diff --git a/charts/skypilot/templates/oauth2-proxy-redis.yaml b/charts/skypilot/templates/oauth2-proxy-redis.yaml
new file mode 100644
index 000000000..8d0d68c0b
--- /dev/null
+++ b/charts/skypilot/templates/oauth2-proxy-redis.yaml
@@ -0,0 +1,52 @@
+{{- if and .Values.ingress.enabled (index .Values.ingress "oauth2-proxy" "enabled") (eq (index .Values.ingress "oauth2-proxy" "session-store-type" | default "redis") "redis") (not (index .Values.ingress "oauth2-proxy" "redis-url")) -}}
+{{- $redis := index .Values.ingress "oauth2-proxy" "redis" -}}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ .Release.Name }}-oauth2-proxy-redis
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ .Release.Name }}-oauth2-proxy-redis
+    skypilot.co/component: oauth2-proxy-redis
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: {{ .Release.Name }}-oauth2-proxy-redis
+  template:
+    metadata:
+      labels:
+        app: {{ .Release.Name }}-oauth2-proxy-redis
+        skypilot.co/component: oauth2-proxy-redis
+    spec:
+      containers:
+      - name: redis
+        image: redis:7-alpine
+        ports:
+        - containerPort: 6379
+        resources:
+          requests:
+            cpu: 100m
+            memory: 128Mi
+          limits:
+            cpu: 200m
+            memory: 256Mi
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ .Release.Name }}-oauth2-proxy-redis
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ .Release.Name }}-oauth2-proxy-redis
+    skypilot.co/component: oauth2-proxy-redis
+spec:
+  type: ClusterIP
+  ports:
+  - port: 6379
+    targetPort: 6379
+    protocol: TCP
+    name: redis
+  selector:
+    app: {{ .Release.Name }}-oauth2-proxy-redis
+{{- end -}}
diff --git a/charts/skypilot/templates/oauth2-proxy-service.yaml b/charts/skypilot/templates/oauth2-proxy-service.yaml
new file mode 100644
index 000000000..b8b55bd54
--- /dev/null
+++ b/charts/skypilot/templates/oauth2-proxy-service.yaml
@@ -0,0 +1,18 @@
+{{- if and .Values.ingress.enabled (index .Values.ingress "oauth2-proxy" "enabled") }}
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    app: {{ .Release.Name }}-oauth2-proxy
+    skypilot.co/component: oauth2-proxy
+  name: {{ .Release.Name }}-oauth2-proxy
+  namespace: {{ .Release.Namespace }}
+spec:
+  ports:
+  - name: http
+    port: 4180
+    protocol: TCP
+    targetPort: 4180
+  selector:
+    app: {{ .Release.Name }}-oauth2-proxy
+{{- end }}
diff --git a/charts/skypilot/values.yaml b/charts/skypilot/values.yaml
index f09fb2532..e9d416dd8 100644
--- a/charts/skypilot/values.yaml
+++ b/charts/skypilot/values.yaml
@@ -100,6 +100,32 @@ ingress:
   # Deprecated: use ingress-nginx.controller.service.nodePorts.https instead
   httpsNodePort: 30051 # Set to null to automatically assign a random port
 
+  # OAuth2 Proxy configuration for authentication
+  oauth2-proxy:
+    # Enable/disable OAuth2 Proxy deployment
+    enabled: false
+    # Docker image for OAuth2 Proxy
+    image: "quay.io/oauth2-proxy/oauth2-proxy:v7.9.0"
+    # Okta domain URL for OIDC issuer (required when enabled=true)
+    oidc-issuer-url: null
+    # Client ID from OIDC provider (Okta) (required when enabled=true)
+    client-id: ""
+    # Client Secret from OIDC provider (Okta) (required when enabled=true)
+    client-secret: ""
+    # Use HTTPS (set to true for production environments)
+    use-https: false
+    # Email domains to allow (use "*" for all domains)
+    email-domain: "*"
+    # Session storage settings ("cookie" or "redis")
+    session-store-type: "redis"
+    # Redis connection URL (redis://host[:port][/db-number])
+    # If left as null and session-store-type is redis, a Redis instance will be deployed automatically
+    # If set, this external Redis instance will be used
+    redis-url: null
+    # Cookie timing settings (in seconds)
+    cookie-refresh: null  # refresh tokens after this duration (Access token lifespan minus 1 min)
+    cookie-expire: null  # cookie expiration time (should match refresh token lifespan)
+
 ingress-nginx:
   enabled: true
   controller:
diff --git a/docs/source/examples/auto-failover.rst b/docs/source/examples/auto-failover.rst
index b01988f01..591783b52 100644
--- a/docs/source/examples/auto-failover.rst
+++ b/docs/source/examples/auto-failover.rst
@@ -23,6 +23,50 @@ searching for regions (or clouds) that can provide the requested resources.
   If specific :code:`cloud`, ``region``, or ``zone`` are requested for a
   task, auto-failover retries only within the specified location.
 
+Specifying infra choices
+------------------------
+
+The :code:`resources.infra` field in the SkyPilot YAML can be used to specify
+the infrastructure choice(s) to use when provisioning any compute.
+
+Schema:
+
+- ``<cloud>/<region>/<zone>`` (region and zone are optional)
+- ``k8s/<context-name>`` (context-name is optional)
+- Wildcards are supported in any component.
+
+Example values:
+
+.. code-block:: yaml
+
+  resources:
+    infra: aws  # Use any available AWS region/zone.
+    infra: aws/us-east-1  # Use a particular region.
+    infra: aws/us-east-1/us-east-1a  # Use a particular zone.
+    infra: aws/*/us-east-1a  # Same as above.
+
+    infra: k8s  # Use any available Kubernetes context.
+    infra: k8s/*  # Same as above.
+    infra: k8s/my-cluster-context  # Use a particular Kubernetes context.
+
+See :ref:`yaml-spec-resources-infra` for more details.
+
+If you are using CLI commands, the ``--infra`` argument accepts the same schema:
+
+.. code-block:: console
+
+  sky launch --infra aws
+  sky launch --infra aws/us-east-1
+  sky launch --infra aws/us-east-1/us-east-1a
+  sky launch --infra 'aws/*/us-east-1a'
+  sky launch --infra k8s
+  sky launch --infra 'k8s/*'
+  sky launch --infra k8s/my-cluster-context
+
+See :ref:`cli` for all the commands that accept ``--infra``.
+
+
+
 Provisioning GPUs
 ----------------------
 
diff --git a/docs/source/overview.rst b/docs/source/overview.rst
index d0adcd76f..d0dca0637 100644
--- a/docs/source/overview.rst
+++ b/docs/source/overview.rst
@@ -227,6 +227,54 @@ SkyPilot supports most major cloud providers. See :ref:`cloud-account-setup` for
 
 By default, SkyPilot reuses your existing cloud authentication methods.  Optionally, you can also :ref:`set up <cloud-permissions>` specific roles, permissions, or service accounts for SkyPilot to use.
 
+Example usage:
+
+.. tab-set::
+
+    .. tab-item:: CLI
+        :sync: cli
+
+        .. code-block:: console
+
+            $ sky launch --infra aws  # Use any available region/zone within this cloud.
+            $ sky launch --infra gcp
+            $ sky launch --infra azure
+            $ sky launch --infra nebius
+            $ sky launch --infra runpod
+            $ sky launch --infra lambda
+            # ... Or any other supported cloud.
+
+            # Use a particular region.
+            $ sky launch --infra aws/us-east-1
+            $ sky launch --infra gcp/us-central1
+
+            # Use a particular zone.
+            $ sky launch --infra aws/us-east-1/us-east-1a
+
+    .. tab-item:: Python
+        :sync: python
+
+        .. code-block:: python
+
+            import sky
+            task = sky.Task().set_resources(sky.Resources(
+                infra='aws',  # Use any available region/zone within this cloud.
+                # infra='gcp',
+                # infra='azure',
+                # infra='nebius',
+                # infra='runpod',
+                # infra='lambda',
+                # ... Or any other supported cloud.
+
+                # Use a particular region:
+                # infra='aws/us-east-1',
+                # infra='gcp/us-central1',
+
+                # Use a particular zone:
+                # infra='aws/us-east-1/us-east-1a',
+            ))
+            sky.launch(task)
+
 .. _concept-kubernetes-clusters:
 
 Kubernetes clusters
@@ -246,6 +294,31 @@ between multiple clusters is also supported.
     :align: center
     :class: no-scaled-link, only-dark
 
+Example usage:
+
+.. tab-set::
+
+    .. tab-item:: CLI
+        :sync: cli
+
+        .. code-block:: console
+
+            $ sky launch --infra k8s  # Use any available Kubernetes context.
+            $ sky launch --infra k8s/my-cluster-context  # Use a particular context.
+
+    .. tab-item:: Python
+        :sync: python
+
+        .. code-block:: python
+
+            import sky
+            task = sky.Task().set_resources(sky.Resources(
+                infra='k8s',  # Use any available Kubernetes context.
+                # infra='k8s/my-cluster-context',  # Use a particular context.
+            ))
+            sky.launch(task)
+
+
 See :ref:`kubernetes-overview`.
 
 .. _concept-existing-machines:
diff --git a/docs/source/reference/api-server/api-server-admin-deploy.rst b/docs/source/reference/api-server/api-server-admin-deploy.rst
index 3c2c0e3ae..49101f16d 100644
--- a/docs/source/reference/api-server/api-server-admin-deploy.rst
+++ b/docs/source/reference/api-server/api-server-admin-deploy.rst
@@ -66,6 +66,8 @@ Install the SkyPilot Helm chart with the following command:
 
     For more details on the available configuration options, refer to :ref:`SkyPilot API Server Helm Chart Values <helm-values-spec>`.
 
+The above command will install a SkyPilot API server and ingress-nginx controller in the given namespace, which by default conflicts with other installations. To deploy multiple API servers, refer to :ref:`Reusing ingress-nginx controller for API server <sky-api-server-helm-multiple-deploy>`.
+
 .. tip::
 
     The API server deployed will be configured to use the hosting Kubernetes cluster to launch tasks by default. Refer to :ref:`sky-api-server-configure-credentials` to configure credentials for more clouds and Kubernetes clusters.
@@ -130,7 +132,8 @@ Our default of using a NodePort service is the recommended way to expose the API
 
             $ NODE_PORT=$(kubectl get svc ${RELEASE_NAME}-ingress-controller-np --namespace $NAMESPACE -o jsonpath='{.spec.ports[?(@.name=="http")].nodePort}')
             $ NODE_IP=$(kubectl get nodes -o jsonpath='{ $.items[0].status.addresses[?(@.type=="ExternalIP")].address }')
-            $ ENDPOINT=http://${WEB_USERNAME}:${WEB_PASSWORD}@${NODE_IP}:${NODE_PORT}
+            $ HOST=${NODE_IP}:${NODE_PORT}
+            $ ENDPOINT=http://${WEB_USERNAME}:${WEB_PASSWORD}@${HOST}
             $ echo $ENDPOINT
             http://skypilot:yourpassword@1.1.1.1:30050
 
@@ -636,6 +639,66 @@ If you are upgrading from an early 0.8.0 nightly with a previously deployed Node
 
     Refer to :ref:`sky-get-api-server-url` for how to customize and/or connect to the new service.
 
+.. _sky-api-server-helm-multiple-deploy:
+
+Reusing ingress controller for API server
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+By default, the SkyPilot helm chart will deploy a new ingress-nginx controller when installing the API server. However, the ingress-nginx controller has some cluster-scope resources that will cause conflicts between multiple installations by default. It is recommended to reuse an existing ingress controller if you want to deploy multiple API servers in the same Kubernetes cluster.
+
+To reuse an existing ingress controller, you can set :ref:`ingress-nginx.enabled <helm-values-ingress-nginx-enabled>` to ``false`` and set :ref:`ingress.path <helm-values-ingress-path>` to a unique path for the deploying API server. For example:
+
+.. code-block:: bash
+
+    # The first API server, with niginx-ingress controller deployed
+    # It is assumed that the first API server is already deployed. If it is not deployed yet,
+    # add neccessary values instead of specifying --reuse-values
+    helm upgrade --install $RELEASE_NAME skypilot/skypilot-nightly --devel \
+        --namespace $NAMESPACE \
+        --reuse-values \
+        --set ingress.path=/first-server
+    
+    # The second API server, reusing the existing ingress controller and using a different path
+    ANOTHER_RELEASE_NAME=skypilot2
+    ANOTHER_NAMESPACE=skypilot2
+    # Replace with your username and password to configure the basic auth credentials for the second API server
+    ANOTHER_WEB_USERNAME=skypilot
+    ANOTHER_WEB_PASSWORD=yourpassword2
+    ANOTHER_AUTH_STRING=$(htpasswd -nb $ANOTHER_WEB_USERNAME $ANOTHER_WEB_PASSWORD)
+    # Deploy the API server, either in the same namespace or a different namespace
+    helm upgrade --install $ANOTHER_RELEASE_NAME skypilot/skypilot-nightly --devel \
+        --namespace $ANOTHER_NAMESPACE \
+        --set ingress-nginx.enabled=false \
+        --set ingress.path=/second-server \
+        --set ingress.authCredentials=$ANOTHER_AUTH_STRING
+
+With the above commands, these two API servers will share the same ingress controller and serves under different paths of the same host. To get the endpoints, follow :ref:`Step 2: Get the API server URL <sky-get-api-server-url>` to get the host from the helm release that has the ingress-nginx controller deployed, and then append the basic auth and path to the host:
+
+.. code-block:: bash
+
+    # HOST was the ingress host we got from Step 2
+    $ FIRST_PATH=$(kubectl get ingress ${RELEASE_NAME}-ingress --namespace $NAMESPACE -o jsonpath='{.metadata.annotations.skypilot\.co\/ingress-path}')
+    $ FIRST_ENDPOINT=http://${WEB_USERNAME}:${WEB_PASSWORD}@${HOST}${FIRST_PATH}
+    $ SECOND_PATH=$(kubectl get ingress ${ANOTHER_RELEASE_NAME}-ingress --namespace $ANOTHER_NAMESPACE -o jsonpath='{.metadata.annotations.skypilot\.co\/ingress-path}')
+    $ SECOND_ENDPOINT=http://${ANOTHER_WEB_USERNAME}:${ANOTHER_WEB_PASSWORD}@${HOST}${SECOND_PATH}
+    $ echo $FIRST_ENDPOINT
+    http://skypilot:yourpassword@1.1.1.1/first-server
+    $ echo $SECOND_ENDPOINT
+    http://skypilot:yourpassword2@1.1.1.1/second-server
+
+The same approach also applies when you have a ingress-nginx controller deployed before installing the SkyPilot API server:
+
+.. code-block:: bash
+
+    # The first API server, disabling the ingress-nginx controller to reuse the existing one
+    helm upgrade --install $RELEASE_NAME skypilot/skypilot-nightly --devel \
+        --namespace $NAMESPACE \
+        --set ingress-nginx.enabled=false \
+        --set ingress.path=/skypilot
+
+It is a good practice to specify a unique :ref:`ingress.path <helm-values-ingress-path>` too in this case, to avoid conflicts with other backends hosted on the same ingress controller.
+
+
 .. _sky-api-server-cloud-deploy:
 
 Alternative: Deploy on cloud VMs
diff --git a/docs/source/reference/api-server/api-server-troubleshooting.rst b/docs/source/reference/api-server/api-server-troubleshooting.rst
index d862fef6f..44ce14c4e 100644
--- a/docs/source/reference/api-server/api-server-troubleshooting.rst
+++ b/docs/source/reference/api-server/api-server-troubleshooting.rst
@@ -40,3 +40,17 @@ If the pending reason is ``FailedScheduling`` and the information indicates ther
 .. note::
 
     API server requires at least 4 CPU cores and 8 GiB memory. Setting lower values may cause degraded performance.
+
+.. _sky-api-server-troubleshooting-ingress-nginx-conflicts:
+
+Ingress-nginx controller conflicts
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+If you encounter similar errors as below when deploying the API server:
+
+.. code-block:: console
+
+    Error: Unable to continue with install: IngressClass "nginx" in namespace "" exists and cannot be imported into the current release: invalid ownership metadata; annotation validation error: key "meta.helm.sh/release-name" must equal "test": current value is "skypilot"
+
+It is likely you have an existing ingress-nginx controller deployed in the Kubernetes cluster. Or have an existing SkyPilot API server deployed which by default installs an ingress-nginx controller. If this is the case, you can follow the instructions in :ref:`Reusing ingress controller for API server <sky-api-server-helm-multiple-deploy>` to reuse the existing ingress-nginx controller to solve this issue.
+
diff --git a/docs/source/reference/api-server/examples/api-server-auth-proxy.rst b/docs/source/reference/api-server/examples/api-server-auth-proxy.rst
index 0ef14ed66..0c40ffab4 100644
--- a/docs/source/reference/api-server/examples/api-server-auth-proxy.rst
+++ b/docs/source/reference/api-server/examples/api-server-auth-proxy.rst
@@ -99,8 +99,6 @@ Set up the environment variables for your API server deployment. ``NAMESPACE`` a
 
     NAMESPACE=skypilot # TODO: change to your installed namespace
     RELEASE_NAME=skypilot # TODO: change to your installed release name
-    VERSION=1.0.0-dev20250410 # TODO: change to the version you want to upgrade to
-    IMAGE_REPO=berkeleyskypilot/skypilot-nightly
 
 Use ``helm upgrade`` to redeploy the API server helm chart with the ``skypilot-oauth2-proxy`` deployment. Replace ``<CLIENT ID>`` and ``<CLIENT SECRET>`` with the values from the Okta admin console above, and ``<OKTA URL>`` with your Okta login URL.
 
@@ -113,8 +111,6 @@ Use ``helm upgrade`` to redeploy the API server helm chart with the ``skypilot-o
       --set ingress.oauth2-proxy.client-id=<CLIENT ID> \
       --set ingress.oauth2-proxy.client-secret=<CLIENT SECRET>
 
-If your API server endpoint only has HTTP, not HTTPS, add ``--set ingress.oauth2-proxy.use-http=true``.
-
 To make sure it's working, visit your endpoint URL in a browser. You should be redirected to Okta to sign in.
 
 Now, you can use ``sky api login -e <ENDPOINT>`` to go though the login flow for the CLI.
diff --git a/docs/source/reference/api-server/helm-values-spec.rst b/docs/source/reference/api-server/helm-values-spec.rst
index 79be55e93..815069deb 100644
--- a/docs/source/reference/api-server/helm-values-spec.rst
+++ b/docs/source/reference/api-server/helm-values-spec.rst
@@ -19,7 +19,7 @@ The SkyPilot API server helm chart provides typical `helm values <https://helm.s
 * ``--set``: Specify overrides on the command line.
 
   .. code-block:: bash
-      
+
       helm install $RELEASE_NAME skypilot/skypilot-nightly --set apiService.image="berkeleyskypilot/skypilot:0.9.1"
 
 Values
@@ -69,6 +69,20 @@ Below is the available helm value keys and the default value of each key:
     :ref:`authSecret <helm-values-ingress-authSecret>`: null
     :ref:`authCredentials <helm-values-ingress-authCredentials>`: "username:$apr1$encrypted_password"
     :ref:`path <helm-values-ingress-path>`: '/'
+    :ref:`oauth2-proxy <helm-values-ingress-oauth2-proxy>`:
+      :ref:`enabled <helm-values-ingress-oauth2-proxy-enabled>`: false
+      # Required when enabled:
+      :ref:`oidc-issuer-url <helm-values-ingress-oauth2-proxy-oidc-issuer-url>`: null
+      :ref:`client-id <helm-values-ingress-oauth2-proxy-client-id>`: ""
+      :ref:`client-secret <helm-values-ingress-oauth2-proxy-client-secret>`: ""
+      # Optional settings:
+      :ref:`image <helm-values-ingress-oauth2-proxy-image>`: "quay.io/oauth2-proxy/oauth2-proxy:v7.9.0"
+      :ref:`use-https <helm-values-ingress-oauth2-proxy-use-https>`: false
+      :ref:`email-domain <helm-values-ingress-oauth2-proxy-email-domain>`: "*"
+      :ref:`session-store-type <helm-values-ingress-oauth2-proxy-session-store-type>`: "redis"
+      :ref:`redis-url <helm-values-ingress-oauth2-proxy-redis-url>`: null
+      :ref:`cookie-refresh <helm-values-ingress-oauth2-proxy-cookie-refresh>`: null
+      :ref:`cookie-expire <helm-values-ingress-oauth2-proxy-cookie-expire>`: null
 
   :ref:`ingress-nginx <helm-values-ingress-nginx>`:
     :ref:`enabled <helm-values-ingress-nginx-enabled>`: true
@@ -380,9 +394,9 @@ Default: ``true``
 ``ingress.authSecret``
 ^^^^^^^^^^^^^^^^^^^^^^
 
-Name of the Kubernetes secret containing basic auth credentials for ingress. If not specified, a new secret will be created using ``authCredentials``.
+Name of the Kubernetes secret containing basic auth credentials for ingress. If not specified, a new secret will be created using ``authCredentials``. This is ignored if ``ingress.oauth2-proxy.enabled`` is ``true``.
 
-One of ``ingress.authSecret`` or ``ingress.authCredentials`` must be set.
+One of ``ingress.authSecret`` or ``ingress.authCredentials`` must be set, unless ``ingress.oauth2-proxy.enabled`` is ``true``.
 
 Default: ``null``
 
@@ -396,9 +410,9 @@ Default: ``null``
 ``ingress.authCredentials``
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-Basic auth credentials in the format ``username:encrypted_password``. Used only if ``authSecret`` is not set.
+Basic auth credentials in the format ``username:encrypted_password``. Used only if ``authSecret`` is not set. This is ignored if ``ingress.oauth2-proxy.enabled`` is ``true``.
 
-One of ``ingress.authSecret`` or ``ingress.authCredentials`` must be set.
+One of ``ingress.authSecret`` or ``ingress.authCredentials`` must be set, unless ``ingress.oauth2-proxy.enabled`` is ``true``.
 
 Default: ``"username:$apr1$encrypted_password"``
 
@@ -421,6 +435,200 @@ Default: ``'/'``
   ingress:
     path: '/'
 
+.. _helm-values-ingress-oauth2-proxy:
+
+``ingress.oauth2-proxy``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Configuration for the OAuth2 Proxy authentication for the API server. This enables SSO providers like Okta.
+
+If enabled, ``ingress.authSecret`` and ``ingress.authCredentials`` are ignored.
+
+Default: see the yaml below.
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      enabled: false
+      # Required when enabled:
+      oidc-issuer-url: null
+      client-id: ""
+      client-secret: ""
+      # Optional settings:
+      image: "quay.io/oauth2-proxy/oauth2-proxy:v7.9.0"
+      use-https: false
+      email-domain: "*"
+      session-store-type: "redis"
+      redis-url: null
+      cookie-refresh: null
+      cookie-expire: null
+
+.. _helm-values-ingress-oauth2-proxy-enabled:
+
+``ingress.oauth2-proxy.enabled``
+''''''''''''''''''''''''''''''''''''
+
+Enable OAuth2 Proxy for authentication. When enabled, this will deploy an OAuth2 Proxy component and configure the ingress to use it for authentication instead of basic auth.
+
+Default: ``false``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      enabled: true
+
+.. _helm-values-ingress-oauth2-proxy-oidc-issuer-url:
+
+``ingress.oauth2-proxy.oidc-issuer-url``
+''''''''''''''''''''''''''''''''''''''''
+
+The URL of the OIDC issuer (e.g., your Okta domain). Required when oauth2-proxy is enabled.
+
+Default: ``null``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      oidc-issuer-url: "https://mycompany.okta.com"
+
+.. _helm-values-ingress-oauth2-proxy-client-id:
+
+``ingress.oauth2-proxy.client-id``
+''''''''''''''''''''''''''''''''''
+
+The OAuth client ID from your OIDC provider (e.g., Okta). Required when oauth2-proxy is enabled.
+
+Default: ``""``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      client-id: "0abc123def456"
+
+.. _helm-values-ingress-oauth2-proxy-client-secret:
+
+``ingress.oauth2-proxy.client-secret``
+'''''''''''''''''''''''''''''''''''''''''
+
+The OAuth client secret from your OIDC provider (e.g., Okta). Required when oauth2-proxy is enabled.
+
+Default: ``""``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      client-secret: "abcdef123456"
+
+.. _helm-values-ingress-oauth2-proxy-image:
+
+``ingress.oauth2-proxy.image``
+''''''''''''''''''''''''''''''
+
+Docker image for the OAuth2 Proxy component.
+
+Default: ``"quay.io/oauth2-proxy/oauth2-proxy:v7.9.0"``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      image: "quay.io/oauth2-proxy/oauth2-proxy:v7.9.0"
+
+.. _helm-values-ingress-oauth2-proxy-use-https:
+
+``ingress.oauth2-proxy.use-https``
+''''''''''''''''''''''''''''''''''
+
+Set to ``true`` when using HTTPS for the API server endpoint. When set to ``false``, secure cookies are disabled, which is required for HTTP endpoints.
+
+Default: ``false``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      use-https: true
+
+.. _helm-values-ingress-oauth2-proxy-email-domain:
+
+``ingress.oauth2-proxy.email-domain``
+'''''''''''''''''''''''''''''''''''''''
+
+Email domains to allow for authentication. Use ``"*"`` to allow all email domains.
+
+Default: ``"*"``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      email-domain: "mycompany.com"
+
+.. _helm-values-ingress-oauth2-proxy-session-store-type:
+
+``ingress.oauth2-proxy.session-store-type``
+'''''''''''''''''''''''''''''''''''''''''''
+
+Session storage type for OAuth2 Proxy. Can be set to ``"cookie"`` or ``"redis"``. Using Redis as a session store results in smaller cookies and better performance for large-scale deployments.
+
+Default: ``"redis"``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      session-store-type: "redis"
+
+.. _helm-values-ingress-oauth2-proxy-redis-url:
+
+``ingress.oauth2-proxy.redis-url``
+''''''''''''''''''''''''''''''''''
+
+URL to connect to an external Redis instance for session storage. If set to ``null`` and ``session-store-type`` is ``"redis"``, a Redis instance will be automatically deployed. Format: ``redis://host[:port][/db-number]``
+
+Default: ``null``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      redis-url: "redis://redis-host:6379/0"
+
+.. _helm-values-ingress-oauth2-proxy-cookie-refresh:
+
+``ingress.oauth2-proxy.cookie-refresh``
+'''''''''''''''''''''''''''''''''''''''
+
+Duration in seconds after which to refresh the access token. This should typically be set to the access token lifespan minus 1 minute. If not set, tokens will not be refreshed automatically.
+
+Default: ``null``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      cookie-refresh: 3540  # 59 minutes (for a 60-minute access token)
+
+.. _helm-values-ingress-oauth2-proxy-cookie-expire:
+
+``ingress.oauth2-proxy.cookie-expire``
+''''''''''''''''''''''''''''''''''''''
+
+Expiration time for cookies in seconds. Should match the refresh token lifespan from your OIDC provider.
+
+Default: ``null``
+
+.. code-block:: yaml
+
+  ingress:
+    oauth2-proxy:
+      cookie-expire: 86400  # 24 hours
+
 .. _helm-values-ingress-nginx:
 
 ``ingress-nginx``
diff --git a/docs/source/reference/kubernetes/kubernetes-priorities.rst b/docs/source/reference/kubernetes/kubernetes-priorities.rst
index f7911ed9a..8fac486b7 100644
--- a/docs/source/reference/kubernetes/kubernetes-priorities.rst
+++ b/docs/source/reference/kubernetes/kubernetes-priorities.rst
@@ -70,7 +70,7 @@ We use two simple counter jobs in this example:
 
   # high-priority-job.yaml
   resources:
-    infra: kubernetes
+    infra: k8s
     cpus: 4
 
   run: |
@@ -91,7 +91,7 @@ We use two simple counter jobs in this example:
 
   # low-priority-job.yaml
   resources:
-    infra: kubernetes
+    infra: k8s
     cpus: 4
 
   run: |
diff --git a/docs/source/reference/yaml-spec.rst b/docs/source/reference/yaml-spec.rst
index b800227ac..b954737e6 100644
--- a/docs/source/reference/yaml-spec.rst
+++ b/docs/source/reference/yaml-spec.rst
@@ -26,8 +26,7 @@ Below is the configuration syntax and some example values.  See details under ea
   :ref:`num_nodes <yaml-spec-num-nodes>`: 4
 
   :ref:`resources <yaml-spec-resources>`:
-    # Infra to use. For example: ``aws``, ``aws/us-east-1``, ``kubernetes``,
-    # or, ``kubernetes/my-h100-cluster-context``.
+    # Infra to use. Click to see schema and example values.
     :ref:`infra <yaml-spec-resources-infra>`: aws
 
     # Hardware.
@@ -157,22 +156,27 @@ Per-node resource requirements (optional).
 ~~~~~~~~~~~~~~~~~~~
 
 
-Infrastructure to use (optional). Format: ``<cloud>``, ``<cloud>/<region>``, ``<cloud>/<region>/<zone>``, ``kubernetes/<context-name>``.
+Infrastructure to use (optional).
 
-Examples: ``aws``, ``aws/us-east-1``, ``aws/us-east-1/us-east-1a``, ``aws/*/us-east-1a``, ``kubernetes/my-cluster-context``.
+Schema: ``<cloud>/<region>/<zone>`` (region
+and zone are optional), or ``k8s/<context-name>`` (context-name is optional).
+Wildcards are supported in any component.
+
+Example values: ``aws``, ``aws/us-east-1``, ``aws/us-east-1/us-east-1a``,
+``aws/*/us-east-1a``, ``k8s``, ``k8s/my-cluster-context``.
 
 .. code-block:: yaml
 
   resources:
-    infra: aws
+    infra: aws  # Use any available AWS region/zone.
 
 
 .. code-block:: yaml
 
   resources:
-    infra: kubernetes
+    infra: k8s  # Use any available Kubernetes context.
 
-You can also specify a specific region, zone or kubernetes context.
+You can also specify a specific region, zone, or Kubernetes context.
 
 .. code-block:: yaml
 
@@ -189,7 +193,7 @@ You can also specify a specific region, zone or kubernetes context.
 .. code-block:: yaml
 
   resources:
-    infra: kubernetes/my-h100-cluster-context
+    infra: k8s/my-h100-cluster-context
 
 
 
diff --git a/examples/aws_efa/nccl_efa.yaml b/examples/aws_efa/nccl_efa.yaml
index c73f1c01b..289e21215 100644
--- a/examples/aws_efa/nccl_efa.yaml
+++ b/examples/aws_efa/nccl_efa.yaml
@@ -1,7 +1,7 @@
 name: nccl-test-efa
 
 resources:
-  infra: kubernetes
+  infra: k8s
   accelerators: A100:8
   cpus: 90+
   image_id: docker:public.ecr.aws/hpc-cloud/nccl-tests:latest
diff --git a/sky/check.py b/sky/check.py
index 6663e5087..c7e935137 100644
--- a/sky/check.py
+++ b/sky/check.py
@@ -93,7 +93,7 @@ def check_capabilities(
     ]
 
     combinations = list(itertools.product(clouds_to_check, capabilities))
-    with rich_utils.safe_status('Checking Cloud(s)...'):
+    with rich_utils.safe_status('Checking infra choices...'):
         check_results = subprocess_utils.run_in_parallel(
             check_one_cloud_one_capability, combinations)
 
diff --git a/sky/client/sdk.py b/sky/client/sdk.py
index a21cf70e8..c065e636f 100644
--- a/sky/client/sdk.py
+++ b/sky/client/sdk.py
@@ -10,14 +10,19 @@ Usage example:
     statuses = sky.get(request_id)
 
 """
+import base64
+import binascii
 import getpass
+from http import cookiejar
 import json
 import logging
 import os
 import pathlib
 import subprocess
+import time
 import typing
 from typing import Any, Dict, List, Optional, Tuple, Union
+from urllib import parse as urlparse
 import webbrowser
 
 import click
@@ -1841,7 +1846,73 @@ def api_login(endpoint: Optional[str] = None) -> None:
             not endpoint.startswith('https://')):
         raise click.BadParameter('Endpoint must be a valid URL.')
 
-    server_common.check_server_healthy(endpoint)
+    server_status = server_common.check_server_healthy(endpoint)
+    if server_status == server_common.ApiServerStatus.NEEDS_AUTH:
+        # We detected an auth proxy, so go through the auth proxy cookie flow.
+        parsed_url = urlparse.urlparse(endpoint)
+        token_url = f'{endpoint}/token'
+        click.echo('Authentication is needed. Please visit this URL setup up '
+                   f'the token:{colorama.Style.BRIGHT}\n\n{token_url}'
+                   f'\n{colorama.Style.RESET_ALL}')
+        if webbrowser.open(token_url):
+            click.echo('Opening browser...')
+        token: str = click.prompt('Paste the token')
+
+        # Parse the token.
+        # b64decode will ignore invalid characters, but does some length and
+        # padding checks.
+        try:
+            data = base64.b64decode(token)
+        except binascii.Error as e:
+            raise ValueError(f'Malformed token: {token}') from e
+        logger.debug(f'Token data: {data!r}')
+        try:
+            cookie_dict = json.loads(data)
+        except (json.JSONDecodeError, UnicodeDecodeError) as e:
+            raise ValueError(f'Malformed token data: {data!r}') from e
+        if not isinstance(cookie_dict, dict):
+            raise ValueError(f'Malformed token JSON: {cookie_dict}')
+
+        cookie_jar = cookiejar.MozillaCookieJar()
+        for (name, value) in cookie_dict.items():
+            # dict keys in JSON must be strings
+            assert isinstance(name, str)
+            if not isinstance(value, str):
+                raise ValueError('Malformed token - bad key/value: '
+                                 f'{name}: {value}')
+
+            # See CookieJar._cookie_from_cookie_tuple
+            # oauth2proxy default is Max-Age 604800
+            expires = int(time.time()) + 604800
+            domain = str(parsed_url.hostname)
+            domain_initial_dot = domain.startswith('.')
+            if not domain_initial_dot:
+                domain = '.' + domain
+
+            cookie_jar.set_cookie(
+                cookiejar.Cookie(
+                    version=0,
+                    name=name,
+                    value=value,
+                    port=None,
+                    port_specified=False,
+                    domain=domain,
+                    domain_specified=True,
+                    domain_initial_dot=domain_initial_dot,
+                    path='',
+                    path_specified=False,
+                    secure=False,
+                    expires=expires,
+                    discard=False,
+                    comment=None,
+                    comment_url=None,
+                    rest=dict(),
+                ))
+
+        # Now that the cookies are parsed, save them to the cookie jar.
+        cookie_jar_path = os.path.expanduser(
+            server_common.get_api_cookie_jar_path())
+        cookie_jar.save(cookie_jar_path)
 
     # Set the endpoint in the config file
     config_path = pathlib.Path(
diff --git a/sky/clouds/service_catalog/__init__.py b/sky/clouds/service_catalog/__init__.py
index 88cc3a882..59618e159 100644
--- a/sky/clouds/service_catalog/__init__.py
+++ b/sky/clouds/service_catalog/__init__.py
@@ -9,6 +9,8 @@ from sky.clouds.service_catalog.constants import ALL_CLOUDS
 from sky.clouds.service_catalog.constants import CATALOG_DIR
 from sky.clouds.service_catalog.constants import CATALOG_SCHEMA_VERSION
 from sky.clouds.service_catalog.constants import HOSTED_CATALOG_DIR_URL
+from sky.clouds.service_catalog.constants import (
+    HOSTED_CATALOG_DIR_URL_S3_MIRROR)
 from sky.utils import resources_utils
 from sky.utils import subprocess_utils
 
@@ -383,6 +385,7 @@ __all__ = [
     # Constants
     'ALL_CLOUDS',
     'HOSTED_CATALOG_DIR_URL',
+    'HOSTED_CATALOG_DIR_URL_S3_MIRROR',
     'CATALOG_SCHEMA_VERSION',
     'CATALOG_DIR',
 ]
diff --git a/sky/clouds/service_catalog/common.py b/sky/clouds/service_catalog/common.py
index e0b603586..39b2e6700 100644
--- a/sky/clouds/service_catalog/common.py
+++ b/sky/clouds/service_catalog/common.py
@@ -197,6 +197,8 @@ def read_catalog(filename: str,
         with filelock.FileLock(meta_path + '.lock'):
             if _need_update():
                 url = f'{constants.HOSTED_CATALOG_DIR_URL}/{constants.CATALOG_SCHEMA_VERSION}/{filename}'  # pylint: disable=line-too-long
+                url_fallback = f'{constants.HOSTED_CATALOG_DIR_URL_S3_MIRROR}/{constants.CATALOG_SCHEMA_VERSION}/{filename}'  # pylint: disable=line-too-long
+                headers = {'User-Agent': 'SkyPilot/0.7'}
                 update_frequency_str = ''
                 if pull_frequency_hours is not None:
                     update_frequency_str = (
@@ -206,8 +208,13 @@ def read_catalog(filename: str,
                             f'Updating {cloud} catalog: {filename}') +
                         f'{update_frequency_str}'):
                     try:
-                        r = requests.get(url=url,
-                                         headers={'User-Agent': 'SkyPilot/0.7'})
+                        r = requests.get(url=url, headers=headers)
+                        if r.status_code == 429:
+                            # fallback to s3 mirror, github introduced rate
+                            # limit after 2025-05, see
+                            # https://github.com/skypilot-org/skypilot/issues/5438
+                            # for more details
+                            r = requests.get(url=url_fallback, headers=headers)
                         r.raise_for_status()
                     except requests.exceptions.RequestException as e:
                         error_str = (f'Failed to fetch {cloud} catalog '
diff --git a/sky/clouds/service_catalog/constants.py b/sky/clouds/service_catalog/constants.py
index 4dc1d5763..91b993aa9 100644
--- a/sky/clouds/service_catalog/constants.py
+++ b/sky/clouds/service_catalog/constants.py
@@ -1,5 +1,6 @@
 """Constants used for service catalog."""
 HOSTED_CATALOG_DIR_URL = 'https://raw.githubusercontent.com/skypilot-org/skypilot-catalog/master/catalogs'  # pylint: disable=line-too-long
+HOSTED_CATALOG_DIR_URL_S3_MIRROR = 'https://skypilot-catalog.s3.us-east-1.amazonaws.com/catalogs'  # pylint: disable=line-too-long
 CATALOG_SCHEMA_VERSION = 'v7'
 CATALOG_DIR = '~/.sky/catalogs'
 ALL_CLOUDS = ('aws', 'azure', 'gcp', 'ibm', 'lambda', 'scp', 'oci',
diff --git a/sky/core.py b/sky/core.py
index c74b1264d..2f60717bf 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -78,14 +78,12 @@ def optimize(
     # is shown on `sky launch`. The optimizer is also invoked during failover,
     # but we do not apply the admin policy there. We should apply the admin
     # policy in the optimizer, but that will require some refactoring.
-    dag, _ = admin_policy_utils.apply(
-        dag,
-        use_mutated_config_in_current_request=True,
-        request_options=request_options)
-    return optimizer.Optimizer.optimize(dag=dag,
-                                        minimize=minimize,
-                                        blocked_resources=blocked_resources,
-                                        quiet=quiet)
+    with admin_policy_utils.apply_and_use_config_in_current_request(
+            dag, request_options=request_options) as dag:
+        return optimizer.Optimizer.optimize(dag=dag,
+                                            minimize=minimize,
+                                            blocked_resources=blocked_resources,
+                                            quiet=quiet)
 
 
 @usage_lib.entrypoint
diff --git a/sky/execution.py b/sky/execution.py
index b173cc8b4..b0e795a7c 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -170,21 +170,70 @@ def _execute(
       handle: Optional[backends.ResourceHandle]; the handle to the cluster. None
         if dryrun.
     """
-
     dag = dag_utils.convert_entrypoint_to_dag(entrypoint)
     for task in dag.tasks:
         if task.storage_mounts is not None:
             for storage in task.storage_mounts.values():
                 # Ensure the storage is constructed.
                 storage.construct()
-    dag, _ = admin_policy_utils.apply(
-        dag,
-        request_options=admin_policy.RequestOptions(
+    with admin_policy_utils.apply_and_use_config_in_current_request(
+            dag,
+            request_options=admin_policy.RequestOptions(
+                cluster_name=cluster_name,
+                idle_minutes_to_autostop=idle_minutes_to_autostop,
+                down=down,
+                dryrun=dryrun,
+            )) as dag:
+        return _execute_dag(
+            dag,
+            dryrun=dryrun,
+            down=down,
+            stream_logs=stream_logs,
+            handle=handle,
+            backend=backend,
+            retry_until_up=retry_until_up,
+            optimize_target=optimize_target,
+            stages=stages,
             cluster_name=cluster_name,
+            detach_setup=detach_setup,
+            detach_run=detach_run,
             idle_minutes_to_autostop=idle_minutes_to_autostop,
-            down=down,
-            dryrun=dryrun,
-        ))
+            no_setup=no_setup,
+            clone_disk_from=clone_disk_from,
+            skip_unnecessary_provisioning=skip_unnecessary_provisioning,
+            _quiet_optimizer=_quiet_optimizer,
+            _is_launched_by_jobs_controller=_is_launched_by_jobs_controller,
+            _is_launched_by_sky_serve_controller=
+            _is_launched_by_sky_serve_controller)
+
+
+def _execute_dag(
+    dag: 'sky.Dag',
+    dryrun: bool,
+    down: bool,
+    stream_logs: bool,
+    handle: Optional[backends.ResourceHandle],
+    backend: Optional[backends.Backend],
+    retry_until_up: bool,
+    optimize_target: common.OptimizeTarget,
+    stages: Optional[List[Stage]],
+    cluster_name: Optional[str],
+    detach_setup: bool,
+    detach_run: bool,
+    idle_minutes_to_autostop: Optional[int],
+    no_setup: bool,
+    clone_disk_from: Optional[str],
+    skip_unnecessary_provisioning: bool,
+    # pylint: disable=invalid-name
+    _quiet_optimizer: bool,
+    _is_launched_by_jobs_controller: bool,
+    _is_launched_by_sky_serve_controller: bool,
+) -> Tuple[Optional[int], Optional[backends.ResourceHandle]]:
+    """Execute a DAG.
+
+    This is an internal helper function for _execute() and is expected to be
+    called only by _execute().
+    """
     assert len(dag) == 1, f'We support 1 task for now. {dag}'
     task = dag.tasks[0]
 
diff --git a/sky/jobs/server/core.py b/sky/jobs/server/core.py
index e64befc74..c6b6c118b 100644
--- a/sky/jobs/server/core.py
+++ b/sky/jobs/server/core.py
@@ -81,8 +81,7 @@ def launch(
     # Always apply the policy again here, even though it might have been applied
     # in the CLI. This is to ensure that we apply the policy to the final DAG
     # and get the mutated config.
-    dag, mutated_user_config = admin_policy_utils.apply(
-        dag, use_mutated_config_in_current_request=False)
+    dag, mutated_user_config = admin_policy_utils.apply(dag)
     if not dag.is_chain():
         with ux_utils.print_exception_no_traceback():
             raise ValueError('Only single-task or chain DAG is '
@@ -189,6 +188,9 @@ def launch(
             **controller_utils.shared_controller_vars_to_fill(
                 controller,
                 remote_user_config_path=remote_user_config_path,
+                # TODO(aylei): the mutated config will not be updated
+                # afterwards without recreate the controller. Need to
+                # revisit this.
                 local_user_config=mutated_user_config,
             ),
         }
diff --git a/sky/serve/server/core.py b/sky/serve/server/core.py
index 7aeaa0272..5364ae522 100644
--- a/sky/serve/server/core.py
+++ b/sky/serve/server/core.py
@@ -141,8 +141,7 @@ def up(
     # Always apply the policy again here, even though it might have been applied
     # in the CLI. This is to ensure that we apply the policy to the final DAG
     # and get the mutated config.
-    dag, mutated_user_config = admin_policy_utils.apply(
-        task, use_mutated_config_in_current_request=False)
+    dag, mutated_user_config = admin_policy_utils.apply(task)
     task = dag.tasks[0]
 
     with rich_utils.safe_status(
@@ -352,8 +351,7 @@ def update(
     # and get the mutated config.
     # TODO(cblmemo,zhwu): If a user sets a new skypilot_config, the update
     # will not apply the config.
-    dag, _ = admin_policy_utils.apply(
-        task, use_mutated_config_in_current_request=False)
+    dag, _ = admin_policy_utils.apply(task)
     task = dag.tasks[0]
 
     assert task.service is not None
diff --git a/sky/server/common.py b/sky/server/common.py
index 12904d612..8bae38135 100644
--- a/sky/server/common.py
+++ b/sky/server/common.py
@@ -12,7 +12,7 @@ import subprocess
 import sys
 import time
 import typing
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Literal, Optional
 from urllib import parse
 import uuid
 
@@ -116,6 +116,7 @@ class ApiServerStatus(enum.Enum):
     HEALTHY = 'healthy'
     UNHEALTHY = 'unhealthy'
     VERSION_MISMATCH = 'version_mismatch'
+    NEEDS_AUTH = 'needs_auth'
 
 
 @dataclasses.dataclass
@@ -127,15 +128,21 @@ class ApiServerInfo:
     commit: Optional[str] = None
 
 
+def get_api_cookie_jar_path() -> str:
+    return os.environ.get(server_constants.API_COOKIE_FILE_ENV_VAR,
+                          server_constants.API_COOKIE_FILE_DEFAULT_LOCATION)
+
+
 def get_api_cookie_jar() -> requests.cookies.RequestsCookieJar:
     """Returns the cookie jar used by the client to access the API server."""
-    cookie_file = os.environ.get(server_constants.API_COOKIE_FILE_ENV_VAR)
     cookie_jar = requests.cookies.RequestsCookieJar()
-    if cookie_file and os.path.exists(cookie_file):
+    cookie_file = get_api_cookie_jar_path()
+    if cookie_file:
         cookie_path = pathlib.Path(cookie_file).expanduser().resolve()
-        file_cookie_jar = MozillaCookieJar(cookie_path)
-        file_cookie_jar.load()
-        cookie_jar.update(file_cookie_jar)
+        if cookie_path.exists():
+            file_cookie_jar = MozillaCookieJar(cookie_path)
+            file_cookie_jar.load()
+            cookie_jar.update(file_cookie_jar)
     return cookie_jar
 
 
@@ -196,6 +203,7 @@ def get_api_server_status(endpoint: Optional[str] = None) -> ApiServerInfo:
             response = requests.get(f'{server_url}/api/health',
                                     timeout=2.5,
                                     cookies=get_api_cookie_jar())
+            logger.debug(f'Health check status: {response.status_code}')
             if response.status_code == 200:
                 try:
                     result = response.json()
@@ -217,9 +225,24 @@ def get_api_server_status(endpoint: Optional[str] = None) -> ApiServerInfo:
                         server_info.status = ApiServerStatus.VERSION_MISMATCH
                     return server_info
                 except (json.JSONDecodeError, AttributeError) as e:
+                    # Try to check if we got redirected to a login page.
+                    for prev_response in response.history:
+                        logger.debug(f'Previous response: {prev_response.url}')
+                        # Heuristic: check if the url looks like a login page or
+                        # oauth flow.
+                        if any(key in prev_response.url
+                               for key in ['login', 'oauth2']):
+                            logger.debug(
+                                f'URL {prev_response.url} looks like '
+                                'a login page or oauth flow, so try to '
+                                'get the cookie.')
+                            return ApiServerInfo(
+                                status=ApiServerStatus.NEEDS_AUTH)
                     logger.warning('Failed to parse API server response: '
                                    f'{str(e)}')
                     return ApiServerInfo(status=ApiServerStatus.UNHEALTHY)
+            elif response.status_code == 401:
+                return ApiServerInfo(status=ApiServerStatus.NEEDS_AUTH)
             else:
                 return ApiServerInfo(status=ApiServerStatus.UNHEALTHY)
         except requests.exceptions.Timeout:
@@ -297,7 +320,6 @@ def _start_api_server(deploy: bool = False,
 
         log_path = os.path.expanduser(constants.API_SERVER_LOGS)
         os.makedirs(os.path.dirname(log_path), exist_ok=True)
-        cmd = f'{" ".join(args)} > {log_path} 2>&1 < /dev/null'
 
         # Start the API server process in the background and don't wait for it.
         # If this is called from a CLI invocation, we need
@@ -305,10 +327,20 @@ def _start_api_server(deploy: bool = False,
         # the API server.
         server_env = os.environ.copy()
         server_env[constants.ENV_VAR_IS_SKYPILOT_SERVER] = 'true'
-        proc = subprocess.Popen(cmd,
-                                shell=True,
-                                start_new_session=True,
-                                env=server_env)
+        with open(log_path, 'w', encoding='utf-8') as log_file:
+            # Because the log file is opened using a with statement, it may seem
+            # that the file will be closed when the with statement is exited
+            # causing the child process to be unable to write to the log file.
+            # However, Popen makes the file descriptor inheritable which means
+            # the child process will inherit its own copy of the fd,
+            # independent of the parent's fd table which enables to child
+            # process to continue writing to the log file.
+            proc = subprocess.Popen(args,
+                                    stdout=log_file,
+                                    stderr=subprocess.STDOUT,
+                                    stdin=subprocess.DEVNULL,
+                                    start_new_session=True,
+                                    env=server_env)
 
         start_time = time.time()
         while True:
@@ -360,7 +392,12 @@ def _start_api_server(deploy: bool = False,
                 f'SkyPilot API server started. {dashboard_msg}'))
 
 
-def check_server_healthy(endpoint: Optional[str] = None,) -> None:
+def check_server_healthy(
+    endpoint: Optional[str] = None
+) -> Literal[
+        # Use an incomplete list of Literals here to enforce raising for other
+        # enum values.
+        ApiServerStatus.HEALTHY, ApiServerStatus.NEEDS_AUTH]:
     """Check if the API server is healthy.
 
     Args:
@@ -370,6 +407,11 @@ def check_server_healthy(endpoint: Optional[str] = None,) -> None:
     Raises:
         RuntimeError: If the server is not healthy or the client version does
             not match the server version.
+
+    Returns:
+        ApiServerStatus: The status of the API server, unless the server is
+            unhealthy or the client version does not match the server version,
+            in which case an exception is raised.
     """
     endpoint = endpoint if endpoint is not None else get_server_url()
     api_server_info = get_api_server_status(endpoint)
@@ -432,6 +474,8 @@ def check_server_healthy(endpoint: Optional[str] = None,) -> None:
 
         hinted_for_server_install_version_mismatch = True
 
+    return api_server_status
+
 
 def _get_version_info_hint(server_info: ApiServerInfo) -> str:
     assert server_info.version is not None, 'Server version is None'
@@ -482,11 +526,13 @@ def get_skypilot_version_on_disk() -> str:
 def check_server_healthy_or_start_fn(deploy: bool = False,
                                      host: str = '127.0.0.1',
                                      foreground: bool = False):
+    api_server_status = None
     try:
-        check_server_healthy()
+        api_server_status = check_server_healthy()
     except exceptions.ApiServerConnectionError as exc:
         endpoint = get_server_url()
-        if not is_api_server_local():
+        if (not is_api_server_local() or
+                api_server_status == ApiServerStatus.NEEDS_AUTH):
             with ux_utils.print_exception_no_traceback():
                 raise exceptions.ApiServerConnectionError(endpoint) from exc
         # Lock to prevent multiple processes from starting the server at the
diff --git a/sky/server/constants.py b/sky/server/constants.py
index e14fced41..bdeff981a 100644
--- a/sky/server/constants.py
+++ b/sky/server/constants.py
@@ -26,6 +26,8 @@ CLUSTER_REFRESH_DAEMON_INTERVAL_SECONDS = 60
 
 # Environment variable for a file path to the API cookie file.
 API_COOKIE_FILE_ENV_VAR = f'{constants.SKYPILOT_ENV_VAR_PREFIX}API_COOKIE_FILE'
+# Default file if unset.
+API_COOKIE_FILE_DEFAULT_LOCATION = '~/.sky/cookies.txt'
 
 # The path to the dashboard build output
 DASHBOARD_DIR = os.path.join(os.path.dirname(__file__), '..', 'dashboard',
diff --git a/sky/server/html/token_page.html b/sky/server/html/token_page.html
new file mode 100644
index 000000000..2be9e736c
--- /dev/null
+++ b/sky/server/html/token_page.html
@@ -0,0 +1,154 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>SkyPilot API Server Login</title>
+    <style>
+        body {
+            font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
+            display: flex;
+            flex-direction: column;
+            align-items: center;
+            justify-content: center;
+            min-height: 100vh;
+            margin: 0;
+            background-color: #f8f9fa;
+            color: #202124;
+            padding: 20px;
+            box-sizing: border-box;
+        }
+        .container {
+            background-color: #ffffff;
+            padding: 48px;
+            border-radius: 8px;
+            box-shadow: 0 1px 3px rgba(0,0,0,0.12), 0 1px 2px rgba(0,0,0,0.24);
+            text-align: center;
+            max-width: 600px;
+            width: 100%;
+        }
+        .logo {
+            width: 64px;
+            height: 64px;
+            margin-bottom: 20px;
+            display: inline-block;
+        }
+        .logo svg {
+            width: 100%;
+            height: 100%;
+        }
+        h1 {
+            font-size: 24px;
+            font-weight: 500;
+            margin-bottom: 20px;
+            color: #202124;
+        }
+        p {
+            font-size: 14px;
+            line-height: 1.5;
+            margin-bottom: 20px;
+            color: #5f6368;
+        }
+        .code-block {
+            background-color: #f1f3f4;
+            border: 1px solid #dadce0;
+            border-radius: 4px;
+            padding: 16px;
+            margin-top: 24px;
+            margin-bottom: 24px;
+            margin-left: auto;
+            margin-right: auto;
+            text-align: left;
+            word-break: break-all;
+            white-space: pre-wrap;
+            font-family: "SFMono-Regular", Consolas, "Liberation Mono", Menlo, Courier, monospace;
+            font-size: 13px;
+            line-height: 1.4;
+            max-width: 480px;
+        }
+        #token-box { /* Specifically for the token */
+            height: auto;
+            min-height: 6em; /* Ensure it's a reasonable size */
+            max-height: 15em; /* Prevent it from getting too large */
+            overflow-y: auto;
+        }
+        .copy-button {
+            background-color: #1a73e8;
+            color: white;
+            border: none;
+            border-radius: 4px;
+            padding: 10px 24px;
+            font-size: 14px;
+            font-weight: 500;
+            cursor: pointer;
+            transition: background-color 0.3s;
+            margin-top: 10px;
+        }
+        .copy-button:hover {
+            background-color: #287ae6;
+        }
+        .copy-button:active {
+            background-color: #1b66c9;
+        }
+        .footer-text {
+            font-size: 12px;
+            color: #5f6368;
+            margin-top: 30px;
+        }
+    </style>
+</head>
+<body>
+    <div class="container">
+        <div class="logo">
+            <!-- SkyPilot Logo Icon -->
+            <svg viewBox="0 0 50 50" fill="none" xmlns="http://www.w3.org/2000/svg">
+                <path d="M25.1258 30.8274L19.2842 31.6783L33.8316 46.2268L31.492 37.1925L25.1258 30.8274Z" fill="#372F8A"/>
+                <path d="M46.9433 0.000976562L0.719727 13.1148L15.2661 27.6601L16.633 21.3925L10.3728 15.1323L40.183 6.74118C40.183 6.74118 46.102 0.855027 46.9444 0.00203721L46.9433 0.000976562Z" fill="#372F8A"/>
+                <path d="M40.1821 6.74021L31.4922 37.1925L33.8318 46.2257L46.9445 0C46.1022 0.85299 40.1831 6.73915 40.1831 6.73915L40.1821 6.74021Z" fill="#372F8A"/>
+                <path d="M21.3356 25.6089L19.2842 31.6783L25.1258 30.8275L30.3741 16.6011L30.3275 16.617L21.3356 25.6089Z" fill="#195D7F"/>
+                <path d="M16.632 21.3918L15.2651 27.6605L21.3357 25.6091L30.3276 16.6172L16.632 21.3918Z" fill="#39A4DD"/>
+            </svg>
+        </div>
+        <h1>Sign in to SkyPilot CLI</h1>
+        <p>You are seeing this page because a SkyPilot command requires authentication.</p>
+
+        <p>Please copy the following token and paste it into your SkyPilot CLI prompt:</p>
+        <div id="token-box" class="code-block">SKYPILOT_API_SERVER_USER_TOKEN_PLACEHOLDER</div>
+        <button id="copy-btn" class="copy-button">Copy Token</button>
+
+        <p class="footer-text">You can close this tab after copying the token.</p>
+    </div>
+
+    <script>
+        const tokenBox = document.getElementById('token-box');
+        const copyBtn  = document.getElementById('copy-btn');
+
+        function selectToken() {
+            // For <pre> or <div>, create a range to select its content
+            const range = document.createRange();
+            range.selectNodeContents(tokenBox);
+            const sel = window.getSelection();
+            sel.removeAllRanges();
+            sel.addRange(range);
+        }
+
+        // Optional: Select the token when the page loads or when token box is clicked
+        tokenBox.addEventListener('click', selectToken);
+        window.addEventListener('load', selectToken);
+
+        copyBtn.addEventListener('click', () => {
+            selectToken(); // Select the text
+            try {
+                document.execCommand('copy');
+                copyBtn.textContent = 'Copied!';
+            } catch (err) {
+                copyBtn.textContent = 'Error!';
+                console.error('Failed to copy text: ', err);
+            }
+            setTimeout(() => {
+                copyBtn.textContent = 'Copy Token';
+            }, 2000);
+        });
+    </script>
+</body>
+</html>
diff --git a/sky/server/requests/executor.py b/sky/server/requests/executor.py
index 0d17f3b4e..e9993a8a1 100644
--- a/sky/server/requests/executor.py
+++ b/sky/server/requests/executor.py
@@ -20,8 +20,6 @@ See the [README.md](../README.md) for detailed architecture of the executor.
 """
 import asyncio
 import contextlib
-import contextvars
-import functools
 import multiprocessing
 import os
 import queue as queue_lib
@@ -52,6 +50,7 @@ from sky.skylet import constants
 from sky.utils import annotations
 from sky.utils import common_utils
 from sky.utils import context
+from sky.utils import context_utils
 from sky.utils import subprocess_utils
 from sky.utils import timeline
 
@@ -368,10 +367,8 @@ async def execute_request_coroutine(request: api_requests.Request):
     # 1. skypilot config is not contextual
     # 2. envs that read directly from os.environ are not contextual
     ctx.override_envs(request_body.env_vars)
-    loop = asyncio.get_running_loop()
-    pyctx = contextvars.copy_context()
-    func_call = functools.partial(pyctx.run, func, **request_body.to_kwargs())
-    fut: asyncio.Future = loop.run_in_executor(None, func_call)
+    fut: asyncio.Future = context_utils.to_thread(func,
+                                                  **request_body.to_kwargs())
 
     async def poll_task(request_id: str) -> bool:
         request = api_requests.get_request(request_id)
diff --git a/sky/server/server.py b/sky/server/server.py
index 078bd3714..bd57d511b 100644
--- a/sky/server/server.py
+++ b/sky/server/server.py
@@ -2,9 +2,11 @@
 
 import argparse
 import asyncio
+import base64
 import contextlib
 import dataclasses
 import datetime
+import json
 import logging
 import multiprocessing
 import os
@@ -49,6 +51,7 @@ from sky.utils import admin_policy_utils
 from sky.utils import common as common_lib
 from sky.utils import common_utils
 from sky.utils import context
+from sky.utils import context_utils
 from sky.utils import dag_utils
 from sky.utils import env_options
 from sky.utils import status_lib
@@ -218,6 +221,34 @@ app.include_router(jobs_rest.router, prefix='/jobs', tags=['jobs'])
 app.include_router(serve_rest.router, prefix='/serve', tags=['serve'])
 
 
+@app.get('/token')
+async def token(request: fastapi.Request) -> fastapi.responses.HTMLResponse:
+    # Use base64 encoding to avoid having to escape anything in the HTML.
+    json_bytes = json.dumps(request.cookies).encode('utf-8')
+    base64_str = base64.b64encode(json_bytes).decode('utf-8')
+
+    html_dir = pathlib.Path(__file__).parent / 'html'
+    token_page_path = html_dir / 'token_page.html'
+    try:
+        with open(token_page_path, 'r', encoding='utf-8') as f:
+            html_content = f.read()
+    except FileNotFoundError as e:
+        raise fastapi.HTTPException(
+            status_code=500, detail='Token page template not found.') from e
+
+    html_content = html_content.replace(
+        'SKYPILOT_API_SERVER_USER_TOKEN_PLACEHOLDER', base64_str)
+
+    return fastapi.responses.HTMLResponse(
+        content=html_content,
+        headers={
+            'Cache-Control': 'no-cache, no-transform',
+            # X-Accel-Buffering: no is useful for preventing buffering issues
+            # with some reverse proxies.
+            'X-Accel-Buffering': 'no'
+        })
+
+
 @app.post('/check')
 async def check(request: fastapi.Request,
                 check_body: payloads.CheckBody) -> None:
@@ -327,25 +358,26 @@ async def validate(validate_body: payloads.ValidateBody) -> None:
     # pairs.
     logger.debug(f'Validating tasks: {validate_body.dag}')
 
+    context.initialize()
+
     def validate_dag(dag: dag_utils.dag_lib.Dag):
         # TODO: Admin policy may contain arbitrary code, which may be expensive
         # to run and may block the server thread. However, moving it into the
         # executor adds a ~150ms penalty on the local API server because of
         # added RTTs. For now, we stick to doing the validation inline in the
         # server thread.
-        dag, _ = admin_policy_utils.apply(
-            dag, request_options=validate_body.request_options)
-        # Skip validating workdir and file_mounts, as those need to be
-        # validated after the files are uploaded to the SkyPilot API server
-        # with `upload_mounts_to_api_server`.
-        dag.validate(skip_file_mounts=True, skip_workdir=True)
+        with admin_policy_utils.apply_and_use_config_in_current_request(
+                dag, request_options=validate_body.request_options) as dag:
+            # Skip validating workdir and file_mounts, as those need to be
+            # validated after the files are uploaded to the SkyPilot API server
+            # with `upload_mounts_to_api_server`.
+            dag.validate(skip_file_mounts=True, skip_workdir=True)
 
     try:
         dag = dag_utils.load_chain_dag_from_yaml_str(validate_body.dag)
-        loop = asyncio.get_running_loop()
         # Apply admin policy and validate DAG is blocking, run it in a separate
         # thread executor to avoid blocking the uvicorn event loop.
-        await loop.run_in_executor(None, validate_dag, dag)
+        await context_utils.to_thread(validate_dag, dag)
     except Exception as e:  # pylint: disable=broad-except
         raise fastapi.HTTPException(
             status_code=400, detail=exceptions.serialize_exception(e)) from e
diff --git a/sky/skypilot_config.py b/sky/skypilot_config.py
index 80f02eca3..5a20bfc9c 100644
--- a/sky/skypilot_config.py
+++ b/sky/skypilot_config.py
@@ -52,6 +52,7 @@ import contextlib
 import copy
 import json
 import os
+import tempfile
 import threading
 import typing
 from typing import Any, Dict, Iterator, List, Optional, Tuple
@@ -62,6 +63,7 @@ from sky.adaptors import common as adaptors_common
 from sky.skylet import constants
 from sky.utils import common_utils
 from sky.utils import config_utils
+from sky.utils import context
 from sky.utils import schemas
 from sky.utils import ux_utils
 
@@ -105,13 +107,66 @@ ENV_VAR_PROJECT_CONFIG = f'{constants.SKYPILOT_ENV_VAR_PREFIX}PROJECT_CONFIG'
 _GLOBAL_CONFIG_PATH = '~/.sky/config.yaml'
 _PROJECT_CONFIG_PATH = '.sky.yaml'
 
-# The loaded config.
-_dict = config_utils.Config()
-_loaded_config_path: Optional[str] = None
-_config_overridden: bool = False
+
+class ConfigContext:
+
+    def __init__(self,
+                 config: config_utils.Config = config_utils.Config(),
+                 config_path: Optional[str] = None,
+                 config_overridden: bool = False):
+        self.config = config
+        self.config_path = config_path
+        self.config_overridden = config_overridden
+
+
+# The global loaded config.
+_global_config_context = ConfigContext()
 _reload_config_lock = threading.Lock()
 
 
+def _get_config_context() -> ConfigContext:
+    """Get config context for current context.
+
+    If no context is available, the global config context is returned.
+    """
+    ctx = context.get()
+    if not ctx:
+        return _global_config_context
+    if ctx.config_context is None:
+        # Config context for current context is not initialized, inherit from
+        # the global one.
+        ctx.config_context = ConfigContext(
+            config=copy.deepcopy(_global_config_context.config),
+            config_path=_global_config_context.config_path,
+            config_overridden=_global_config_context.config_overridden,
+        )
+    return ctx.config_context
+
+
+def _get_loaded_config() -> config_utils.Config:
+    return _get_config_context().config
+
+
+def _set_loaded_config(config: config_utils.Config) -> None:
+    _get_config_context().config = config
+
+
+def _get_loaded_config_path() -> Optional[str]:
+    return _get_config_context().config_path
+
+
+def _set_loaded_config_path(path: Optional[str]) -> None:
+    _get_config_context().config_path = path
+
+
+def _is_config_overridden() -> bool:
+    return _get_config_context().config_overridden
+
+
+def _set_config_overridden(config_overridden: bool) -> None:
+    _get_config_context().config_overridden = config_overridden
+
+
 def get_user_config_path() -> str:
     """Returns the path to the user config file."""
     return _GLOBAL_CONFIG_PATH
@@ -224,7 +279,7 @@ def get_nested(keys: Tuple[str, ...],
     Returns:
         The value of the nested key, or 'default_value' if not found.
     """
-    return _dict.get_nested(
+    return _get_loaded_config().get_nested(
         keys,
         default_value,
         override_configs,
@@ -237,14 +292,14 @@ def set_nested(keys: Tuple[str, ...], value: Any) -> Dict[str, Any]:
 
     Like get_nested(), if any key is not found, this will not raise an error.
     """
-    copied_dict = copy.deepcopy(_dict)
+    copied_dict = copy.deepcopy(_get_loaded_config())
     copied_dict.set_nested(keys, value)
     return dict(**copied_dict)
 
 
 def to_dict() -> config_utils.Config:
     """Returns a deep-copied version of the current config."""
-    return copy.deepcopy(_dict)
+    return copy.deepcopy(_get_loaded_config())
 
 
 def _get_config_file_path(envvar: str) -> Optional[str]:
@@ -345,10 +400,9 @@ def _parse_dotlist(dotlist: List[str]) -> config_utils.Config:
 
 
 def _reload_config_from_internal_file(internal_config_path: str) -> None:
-    global _dict, _loaded_config_path
     # Reset the global variables, to avoid using stale values.
-    _dict = config_utils.Config()
-    _loaded_config_path = None
+    _set_loaded_config(config_utils.Config())
+    _set_loaded_config_path(None)
 
     config_path = os.path.expanduser(internal_config_path)
     if not os.path.exists(config_path):
@@ -359,14 +413,13 @@ def _reload_config_from_internal_file(internal_config_path: str) -> None:
                 'exist. Please double check the path or unset the env var: '
                 f'unset {ENV_VAR_SKYPILOT_CONFIG}')
     logger.debug(f'Using config path: {config_path}')
-    _dict = parse_config_file(config_path)
-    _loaded_config_path = config_path
+    _set_loaded_config(parse_config_file(config_path))
+    _set_loaded_config_path(config_path)
 
 
 def _reload_config_as_server() -> None:
-    global _dict
     # Reset the global variables, to avoid using stale values.
-    _dict = config_utils.Config()
+    _set_loaded_config(config_utils.Config())
 
     overrides: List[config_utils.Config] = []
     server_config = get_server_config()
@@ -382,13 +435,12 @@ def _reload_config_as_server() -> None:
         logger.debug(
             f'server config: \n'
             f'{common_utils.dump_yaml_str(dict(overlaid_server_config))}')
-    _dict = overlaid_server_config
+    _set_loaded_config(overlaid_server_config)
 
 
 def _reload_config_as_client() -> None:
-    global _dict
     # Reset the global variables, to avoid using stale values.
-    _dict = config_utils.Config()
+    _set_loaded_config(config_utils.Config())
 
     overrides: List[config_utils.Config] = []
     user_config = get_user_config()
@@ -407,15 +459,15 @@ def _reload_config_as_client() -> None:
         logger.debug(
             f'client config (before task and CLI overrides): \n'
             f'{common_utils.dump_yaml_str(dict(overlaid_client_config))}')
-    _dict = overlaid_client_config
+    _set_loaded_config(overlaid_client_config)
 
 
 def loaded_config_path() -> Optional[str]:
     """Returns the path to the loaded config file, or
     '<overridden>' if the config is overridden."""
-    if _config_overridden:
+    if _is_config_overridden():
         return '<overridden>'
-    return _loaded_config_path
+    return _get_loaded_config_path()
 
 
 # Load on import, synchronization is guaranteed by python interpreter.
@@ -424,21 +476,20 @@ _reload_config()
 
 def loaded() -> bool:
     """Returns if the user configurations are loaded."""
-    return bool(_dict)
+    return bool(_get_loaded_config())
 
 
 @contextlib.contextmanager
 def override_skypilot_config(
         override_configs: Optional[Dict[str, Any]]) -> Iterator[None]:
     """Overrides the user configurations."""
-    global _dict, _config_overridden
     # TODO(SKY-1215): allow admin user to extend the disallowed keys or specify
     # allowed keys.
     if not override_configs:
         # If no override configs (None or empty dict), do nothing.
         yield
         return
-    original_config = _dict
+    original_config = _get_loaded_config()
     override_configs = config_utils.Config(override_configs)
     disallowed_diff_keys = []
     for key in constants.SKIPPED_CLIENT_OVERRIDE_KEYS:
@@ -455,7 +506,7 @@ def override_skypilot_config(
             'and will be ignored. Remove these keys to disable this warning. '
             'If you want to specify it, please modify it on server side or '
             'contact your administrator.')
-    config = _dict.get_nested(
+    config = original_config.get_nested(
         keys=tuple(),
         default_value=None,
         override_configs=dict(override_configs),
@@ -469,8 +520,8 @@ def override_skypilot_config(
             'https://docs.skypilot.co/en/latest/reference/config.html. '  # pylint: disable=line-too-long
             'Error: ',
             skip_none=False)
-        _config_overridden = True
-        _dict = config
+        _set_config_overridden(True)
+        _set_loaded_config(config)
         yield
     except exceptions.InvalidSkyPilotConfigError as e:
         with ux_utils.print_exception_no_traceback():
@@ -483,8 +534,43 @@ def override_skypilot_config(
                 f'{common_utils.dump_yaml_str(dict(override_configs))}\n'
                 f'Details: {e}') from e
     finally:
-        _dict = original_config
-        _config_overridden = False
+        _set_loaded_config(original_config)
+        _set_config_overridden(False)
+
+
+@contextlib.contextmanager
+def replace_skypilot_config(new_configs: config_utils.Config) -> Iterator[None]:
+    """Replaces the global config with the new configs.
+
+    This function is concurrent safe when it is:
+    1. called in different processes;
+    2. or called in a same process but with different context, refer to
+       sky_utils.context for more details.
+    """
+    original_config = _get_loaded_config()
+    original_env_var = os.environ.get(ENV_VAR_SKYPILOT_CONFIG)
+    if new_configs != original_config:
+        # Modify the global config of current process or context
+        _set_loaded_config(new_configs)
+        with tempfile.NamedTemporaryFile(delete=False,
+                                         mode='w',
+                                         prefix='mutated-skypilot-config-',
+                                         suffix='.yaml') as temp_file:
+            common_utils.dump_yaml(temp_file.name, dict(**new_configs))
+        # Modify the env var of current process or context so that the
+        # new config will be used by spawned sub-processes.
+        # Note that this code modifies os.environ directly because it
+        # will be hijacked to be context-aware if a context is active.
+        os.environ[ENV_VAR_SKYPILOT_CONFIG] = temp_file.name
+        yield
+        # Restore the original config and env var.
+        _set_loaded_config(original_config)
+        if original_env_var:
+            os.environ[ENV_VAR_SKYPILOT_CONFIG] = original_env_var
+        else:
+            os.environ.pop(ENV_VAR_SKYPILOT_CONFIG, None)
+    else:
+        yield
 
 
 def _compose_cli_config(cli_config: Optional[List[str]]) -> config_utils.Config:
@@ -529,11 +615,11 @@ def apply_cli_config(cli_config: Optional[List[str]]) -> Dict[str, Any]:
         cli_config: A path to a config file or a comma-separated
         list of key-value pairs.
     """
-    global _dict
     parsed_config = _compose_cli_config(cli_config)
     if sky_logging.logging_enabled(logger, sky_logging.DEBUG):
         logger.debug(f'applying following CLI overrides: \n'
                      f'{common_utils.dump_yaml_str(dict(parsed_config))}')
-    _dict = overlay_skypilot_config(original_config=_dict,
-                                    override_configs=parsed_config)
+    _set_loaded_config(
+        overlay_skypilot_config(original_config=_get_loaded_config(),
+                                override_configs=parsed_config))
     return parsed_config
diff --git a/sky/templates/nebius-ray.yml.j2 b/sky/templates/nebius-ray.yml.j2
index 209566bbc..838a176f2 100644
--- a/sky/templates/nebius-ray.yml.j2
+++ b/sky/templates/nebius-ray.yml.j2
@@ -47,11 +47,9 @@ available_node_types:
       ImageId: {{image_id}}
       DiskSize: {{disk_size}}
       UserData: |
-        {%- if docker_image is not none %}
         runcmd:
           - sudo sed -i 's/^#\?AllowTcpForwarding.*/AllowTcpForwarding yes/' /etc/ssh/sshd_config
           - systemctl restart sshd
-        {%- endif %}
 
         {# Two available OS images:
                1. ubuntu22.04-driverless - requires Docker installation
diff --git a/sky/utils/admin_policy_utils.py b/sky/utils/admin_policy_utils.py
index 7b754ed70..3e719c48a 100644
--- a/sky/utils/admin_policy_utils.py
+++ b/sky/utils/admin_policy_utils.py
@@ -1,9 +1,8 @@
 """Admin policy utils."""
+import contextlib
 import copy
 import importlib
-import os
-import tempfile
-from typing import Optional, Tuple, Union
+from typing import Iterator, Optional, Tuple, Union
 
 import colorama
 
@@ -52,9 +51,31 @@ def _get_policy_cls(
     return policy_cls
 
 
+@contextlib.contextmanager
+def apply_and_use_config_in_current_request(
+    entrypoint: Union['dag_lib.Dag', 'task_lib.Task'],
+    request_options: Optional[admin_policy.RequestOptions] = None,
+) -> Iterator['dag_lib.Dag']:
+    """Applies an admin policy and override SkyPilot config for current request
+
+    This is a helper function of `apply()` that applies an admin policy and
+    overrides the SkyPilot config for the current request as a context manager.
+    The original SkyPilot config will be restored when the context manager is
+    exited.
+
+    Refer to `apply()` for more details.
+    """
+    original_config = skypilot_config.to_dict()
+    dag, mutated_config = apply(entrypoint, request_options)
+    if mutated_config != original_config:
+        with skypilot_config.replace_skypilot_config(mutated_config):
+            yield dag
+    else:
+        yield dag
+
+
 def apply(
     entrypoint: Union['dag_lib.Dag', 'task_lib.Task'],
-    use_mutated_config_in_current_request: bool = True,
     request_options: Optional[admin_policy.RequestOptions] = None,
 ) -> Tuple['dag_lib.Dag', config_utils.Config]:
     """Applies an admin policy (if registered) to a DAG or a task.
@@ -85,8 +106,7 @@ def apply(
         return dag, skypilot_config.to_dict()
 
     logger.info(f'Applying policy: {policy}')
-    original_config = skypilot_config.to_dict()
-    config = copy.deepcopy(original_config)
+    config = copy.deepcopy(skypilot_config.to_dict())
     mutated_dag = dag_lib.Dag()
     mutated_dag.name = dag.name
 
@@ -126,22 +146,6 @@ def apply(
         mutated_dag.graph.add_edge(mutated_dag.tasks[u_idx],
                                    mutated_dag.tasks[v_idx])
 
-    if (use_mutated_config_in_current_request and
-            original_config != mutated_config):
-        with tempfile.NamedTemporaryFile(
-                delete=False,
-                mode='w',
-                prefix='policy-mutated-skypilot-config-',
-                suffix='.yaml') as temp_file:
-
-            common_utils.dump_yaml(temp_file.name, dict(**mutated_config))
-            os.environ[skypilot_config.ENV_VAR_SKYPILOT_CONFIG] = temp_file.name
-            logger.debug(f'Updated SkyPilot config: {temp_file.name}')
-            # TODO(zhwu): This is not a clean way to update the SkyPilot config,
-            # because we are resetting the global context for a single DAG,
-            # which is conceptually weird.
-            importlib.reload(skypilot_config)
-
     logger.debug(f'Mutated user request: {mutated_user_request}')
     mutated_dag.policy_applied = True
     return mutated_dag, mutated_config
diff --git a/sky/utils/context.py b/sky/utils/context.py
index 9376e88de..45f19d692 100644
--- a/sky/utils/context.py
+++ b/sky/utils/context.py
@@ -57,6 +57,7 @@ class Context(object):
         self._log_file = None
         self._log_file_handle = None
         self.env_overrides = {}
+        self.config_context = None
 
     def cancel(self):
         """Cancel the context."""
@@ -159,17 +160,25 @@ class ContextualEnviron(MutableMapping):
         ctx = get()
         if ctx is not None:
             if key in ctx.env_overrides:
-                return ctx.env_overrides[key]
+                value = ctx.env_overrides[key]
+                # None is used to indicate that the key is deleted in the
+                # context.
+                if value is None:
+                    raise KeyError(key)
+                return value
         return self._environ[key]
 
     def __iter__(self):
         ctx = get()
+        deleted_keys = set()
         if ctx is not None:
-            for key in ctx.env_overrides:
+            for key, value in ctx.env_overrides.items():
+                if value is None:
+                    deleted_keys.add(key)
                 yield key
             for key in self._environ:
                 # Deduplicate the keys
-                if key not in ctx.env_overrides:
+                if key not in ctx.env_overrides and key not in deleted_keys:
                     yield key
         else:
             return self._environ.__iter__()
@@ -178,10 +187,27 @@ class ContextualEnviron(MutableMapping):
         return len(dict(self))
 
     def __setitem__(self, key, value):
-        return self._environ.__setitem__(key, value)
+        ctx = get()
+        if ctx is not None:
+            ctx.env_overrides[key] = value
+        else:
+            self._environ.__setitem__(key, value)
 
     def __delitem__(self, key):
-        return self._environ.__delitem__(key)
+        ctx = get()
+        if ctx is not None:
+            if key in ctx.env_overrides:
+                del ctx.env_overrides[key]
+            elif key in self._environ:
+                # If the key is not set in the context but set in the environ
+                # of the process, we mark it as deleted in the context by
+                # setting the value to None.
+                ctx.env_overrides[key] = None
+            else:
+                # The key is not set in the context nor the process.
+                raise KeyError(key)
+        else:
+            self._environ.__delitem__(key)
 
     def __repr__(self):
         return self._environ.__repr__()
@@ -190,7 +216,11 @@ class ContextualEnviron(MutableMapping):
         copied = self._environ.copy()
         ctx = get()
         if ctx is not None:
-            copied.update(ctx.env_overrides)
+            for key in ctx.env_overrides:
+                if ctx.env_overrides[key] is None:
+                    copied.pop(key)
+                else:
+                    copied[key] = ctx.env_overrides[key]
         return copied
 
     def setdefault(self, key, default=None):
diff --git a/sky/utils/context_utils.py b/sky/utils/context_utils.py
index 842b86ce1..d6fd7df0b 100644
--- a/sky/utils/context_utils.py
+++ b/sky/utils/context_utils.py
@@ -1,5 +1,6 @@
 """Utilities for SkyPilot context."""
 import asyncio
+import contextvars
 import functools
 import io
 import multiprocessing
@@ -170,3 +171,17 @@ def cancellation_guard(func: F) -> F:
         return func(*args, **kwargs)
 
     return typing.cast(F, wrapper)
+
+
+# TODO(aylei): replace this with asyncio.to_thread once we drop support for
+# python 3.8
+def to_thread(func, /, *args, **kwargs):
+    """Asynchronously run function *func* in a separate thread.
+
+    This is same as asyncio.to_thread added in python 3.9
+    """
+    loop = asyncio.get_running_loop()
+    # This is critical to pass the current coroutine context to the new thread
+    pyctx = contextvars.copy_context()
+    func_call = functools.partial(pyctx.run, func, *args, **kwargs)
+    return loop.run_in_executor(None, func_call)


Test patch:
diff --git a/tests/test_config.py b/tests/test_config.py
index 9e275f8d1..c5f6d240e 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -28,15 +28,14 @@ PROVISION_TIMEOUT = 600
 
 
 def _reload_config() -> None:
-    skypilot_config._dict = config_utils.Config()
-    skypilot_config._loaded_config_path = None
+    skypilot_config._global_config_context = skypilot_config.ConfigContext()
     skypilot_config._reload_config()
 
 
 def _check_empty_config() -> None:
     """Check that the config is empty."""
-    assert not skypilot_config.loaded(), (skypilot_config._dict,
-                                          skypilot_config._loaded_config_path)
+    assert not skypilot_config.loaded(), (
+        skypilot_config._global_config_context)
     assert skypilot_config.get_nested(
         ('aws', 'ssh_proxy_command'), None) is None
     assert skypilot_config.get_nested(('aws', 'ssh_proxy_command'),
@@ -590,8 +589,7 @@ def test_override_skypilot_config_without_original_config(
     monkeypatch.setattr(skypilot_config, '_GLOBAL_CONFIG_PATH', config_path)
     monkeypatch.setattr(skypilot_config, '_PROJECT_CONFIG_PATH', config_path)
     skypilot_config._reload_config()
-    assert not skypilot_config._dict
-    assert skypilot_config._loaded_config_path is None
+    assert not skypilot_config._get_loaded_config()
     assert skypilot_config.get_nested(('aws', 'vpc_name'), None) is None
 
     # Test empty override_configs
@@ -622,8 +620,8 @@ def test_override_skypilot_config_without_original_config(
     assert skypilot_config.get_nested(
         ('aws', 'ssh_proxy_command'), None) is None
     assert os.environ.get(skypilot_config.ENV_VAR_SKYPILOT_CONFIG) is None
-    assert skypilot_config._loaded_config_path is None
-    assert not skypilot_config._dict
+    assert skypilot_config._get_loaded_config_path() is None
+    assert not skypilot_config._get_loaded_config()
 
 
 def test_hierarchical_client_config(monkeypatch, tmp_path):
@@ -706,7 +704,7 @@ def test_hierarchical_client_config(monkeypatch, tmp_path):
     monkeypatch.setattr(skypilot_config, '_PROJECT_CONFIG_PATH',
                         non_existent_config_path)
     skypilot_config._reload_config()
-    assert not skypilot_config._dict
+    assert not skypilot_config._get_loaded_config()
 
     # if config files specified by env vars are missing,
     # error out
diff --git a/tests/test_failover.py b/tests/test_failover.py
index 8b50115e5..041f9a608 100644
--- a/tests/test_failover.py
+++ b/tests/test_failover.py
@@ -11,8 +11,8 @@ import sky
 from sky import global_user_state
 from sky import sky_logging
 from sky.backends import cloud_vm_ray_backend
+from sky.clouds.service_catalog import aws_catalog
 from sky.provision.aws import instance as aws_instance
-from sky.server import config as server_config
 from sky.utils import db_utils
 from sky.utils import env_options
 
@@ -33,6 +33,11 @@ def test_aws_region_failover(enable_all_clouds, _mock_db_conn, mock_aws_backend,
                         lambda: True)
     sky_logging.reload_logger()
 
+    # Ensure AWS catalog dataframes are initialized before mock_aws
+    _ = aws_catalog._default_df._load_df()
+    _ = aws_catalog._image_df._load_df()
+    _ = aws_catalog._quotas_df._load_df()
+
     region_attempt_count = {'count': 0}
 
     def mock_create_instances(ec2_fail_fast, cluster_name, node_config, tags,
diff --git a/tests/unit_tests/test_admin_policy.py b/tests/unit_tests/test_admin_policy.py
index 988ece5aa..770497fe7 100644
--- a/tests/unit_tests/test_admin_policy.py
+++ b/tests/unit_tests/test_admin_policy.py
@@ -34,6 +34,12 @@ def task():
     return sky.Task.from_yaml(os.path.join(POLICY_PATH, 'task.yaml'))
 
 
+def _load_task(task: sky.Task, config_path: str) -> sky.Task:
+    os.environ[skypilot_config.ENV_VAR_SKYPILOT_CONFIG] = config_path
+    importlib.reload(skypilot_config)
+    return task
+
+
 def _load_task_and_apply_policy(
     task: sky.Task,
     config_path: str,
@@ -81,11 +87,11 @@ def test_use_spot_for_all_gpus_policy(add_example_policy_paths, task):
 
 
 def test_add_labels_policy(add_example_policy_paths, task):
-    dag, _ = _load_task_and_apply_policy(
-        task, os.path.join(POLICY_PATH, 'add_labels.yaml'))
-    assert 'app' in skypilot_config.get_nested(
-        ('kubernetes', 'custom_metadata', 'labels'),
-        {}), ('label should be set')
+    task = _load_task(task, os.path.join(POLICY_PATH, 'add_labels.yaml'))
+    with admin_policy_utils.apply_and_use_config_in_current_request(task):
+        assert 'app' in skypilot_config.get_nested(
+            ('kubernetes', 'custom_metadata', 'labels'),
+            {}), ('label should be set')
 
 
 def test_reject_all_policy(add_example_policy_paths, task):
@@ -180,7 +186,7 @@ def test_enforce_autostop_policy(add_example_policy_paths, task):
 @mock.patch('sky.provision.kubernetes.utils.get_all_kube_context_names',
             return_value=['kind-skypilot', 'kind-skypilot2', 'kind-skypilot3'])
 def test_dynamic_kubernetes_contexts_policy(add_example_policy_paths, task):
-    _, config = _load_task_and_apply_policy(
+    dag, config = _load_task_and_apply_policy(
         task,
         os.path.join(POLICY_PATH, 'dynamic_kubernetes_contexts_update.yaml'))
 
@@ -189,7 +195,11 @@ def test_dynamic_kubernetes_contexts_policy(add_example_policy_paths, task):
         None) == ['kind-skypilot', 'kind-skypilot2'
                  ], 'Kubernetes allowed contexts should be updated'
 
+    with admin_policy_utils.apply_and_use_config_in_current_request(dag):
+        assert skypilot_config.get_nested(
+            ('kubernetes', 'allowed_contexts'),
+            None) == ['kind-skypilot', 'kind-skypilot2'
+                     ], 'Global skypilot config should be updated'
     assert skypilot_config.get_nested(
         ('kubernetes', 'allowed_contexts'),
-        None) == ['kind-skypilot',
-                  'kind-skypilot2'], 'Global skypilot config should be updated'
+        None) is None, 'Global skypilot config should be restored after request'
diff --git a/tests/unit_tests/test_backend_utils.py b/tests/unit_tests/test_backend_utils.py
index 459af66e4..9fa2cf6b3 100644
--- a/tests/unit_tests/test_backend_utils.py
+++ b/tests/unit_tests/test_backend_utils.py
@@ -9,8 +9,8 @@ from sky.resources import Resources
 
 
 # Set env var to test config file.
-@mock.patch.object(skypilot_config, '_dict', None)
-@mock.patch.object(skypilot_config, '_loaded_config_path', None)
+@mock.patch.object(skypilot_config, '_global_config_context',
+                   skypilot_config.ConfigContext())
 @mock.patch('sky.clouds.service_catalog.instance_type_exists',
             return_value=True)
 @mock.patch('sky.clouds.service_catalog.get_accelerators_from_instance_type',
diff --git a/tests/unit_tests/test_sky/server/test_server.py b/tests/unit_tests/test_sky/server/test_server.py
index db681b4a8..d4a9ce104 100644
--- a/tests/unit_tests/test_sky/server/test_server.py
+++ b/tests/unit_tests/test_sky/server/test_server.py
@@ -12,6 +12,7 @@ import uvicorn
 
 from sky.server import server
 from sky.utils import common_utils
+from sky.utils import config_utils
 from sky.utils import context
 
 
@@ -113,7 +114,7 @@ async def test_validate():
     with mock.patch('sky.server.server.dag_utils.load_chain_dag_from_yaml_str',
                    return_value=mock_dag), \
          mock.patch('sky.server.server.admin_policy_utils.apply',
-                   return_value=(mock_dag, None)), \
+                   return_value=(mock_dag, config_utils.Config())), \
          mock.patch.object(mock_dag, 'validate') as mock_validate:
         # Call validate endpoint
         await server.validate(mock_validate_body)
@@ -125,7 +126,7 @@ async def test_validate():
     with mock.patch('sky.server.server.dag_utils.load_chain_dag_from_yaml_str',
                    return_value=mock_dag), \
          mock.patch('sky.server.server.admin_policy_utils.apply',
-                   return_value=(mock_dag, None)), \
+                   return_value=(mock_dag, config_utils.Config())), \
          mock.patch.object(mock_dag, 'validate',
                           side_effect=ValueError(error_msg)):
         with pytest.raises(fastapi.HTTPException) as exc_info:
@@ -144,7 +145,7 @@ async def test_validate():
     with mock.patch('sky.server.server.dag_utils.load_chain_dag_from_yaml_str',
                    return_value=mock_dag), \
          mock.patch('sky.server.server.admin_policy_utils.apply',
-                   return_value=(mock_dag, None)), \
+                   return_value=(mock_dag, config_utils.Config())), \
          mock.patch.object(mock_dag, 'validate',
                           side_effect=slow_validate):
         # Start validation in background
diff --git a/tests/unit_tests/test_sky/utils/text_context.py b/tests/unit_tests/test_sky/utils/text_context.py
index a8ba88058..54e4b6e8d 100644
--- a/tests/unit_tests/test_sky/utils/text_context.py
+++ b/tests/unit_tests/test_sky/utils/text_context.py
@@ -4,6 +4,7 @@ import asyncio
 import os
 import pathlib
 import tempfile
+from unittest import mock
 
 import pytest
 
@@ -118,16 +119,17 @@ def test_contextual_environ_set_del(ctx):
 
     # Set a new value in the base environ
     env['NEW_VAR'] = 'direct_value'
-    assert base_environ['NEW_VAR'] == 'direct_value'
+    assert 'NEW_VAR' not in base_environ, 'set env key in context should not affect the process env'
 
     # Delete a value from base environ
     del env['BASE_VAR']
-    assert 'BASE_VAR' not in base_environ
+    assert 'BASE_VAR' in base_environ, 'delete env key in context should not affect the process env'
 
     # Context overrides should take precedence over base
     ctx.override_envs({'NEW_VAR': 'override_value'})
     assert env['NEW_VAR'] == 'override_value'
-    assert base_environ['NEW_VAR'] == 'direct_value'
+    assert base_environ['BASE_VAR'] == 'base_value'
+    assert env.get('BASE_VAR') is None
 
 
 def test_contextual_environ_methods(ctx):