diff --git a/access-control-service/src/main/scala/org/apache/texera/service/resource/AccessControlResource.scala b/access-control-service/src/main/scala/org/apache/texera/service/resource/AccessControlResource.scala
index 792a0dfd8a1..78ab843db4b 100644
--- a/access-control-service/src/main/scala/org/apache/texera/service/resource/AccessControlResource.scala
+++ b/access-control-service/src/main/scala/org/apache/texera/service/resource/AccessControlResource.scala
@@ -30,9 +30,9 @@ import org.apache.texera.auth.util.{ComputingUnitAccess, HeaderField}
 import org.apache.texera.common.config.{GuiConfig, LLMConfig}
 import org.apache.texera.dao.SqlServer
 import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
-import org.apache.texera.dao.jooq.generated.tables.daos.WorkflowComputingUnitDao
+import org.apache.texera.dao.jooq.generated.tables.daos.{UserJupyterDao, WorkflowComputingUnitDao}
 
-import java.net.URLDecoder
+import java.net.{URI, URLDecoder}
 import java.nio.charset.StandardCharsets
 import java.util.Optional
 import scala.jdk.CollectionConverters.{CollectionHasAsScala, MapHasAsScala}
@@ -51,6 +51,10 @@ object AccessControlResource extends LazyLogging {
   private val pvePvesCuidPath: Regex = """^/?(?:auth/)?(?:api/|wsapi/)?pve/pves/([0-9]+)$""".r
   private val pvePackagesCuidPath: Regex =
     """^/?(?:auth/)?(?:api/|wsapi/)?pve/([0-9]+)/[^/]+/packages/.+$""".r
+  // Per-user JupyterLab. The uid is in the path because a browser cannot attach Texera
+  // credentials to the requests Jupyter's own scripts make, so it is the only place the
+  // owner can be read from.
+  private val jupyterPath: Regex = """^/?(?:auth/)?jupyter/([0-9]+)(?:/.*)?$""".r
 
   /**
     * Authorize the request based on the path and headers.
@@ -68,6 +72,7 @@ object AccessControlResource extends LazyLogging {
     logger.info(s"Authorizing request for path: $path")
 
     path match {
+      case jupyterPath(uid) => routeToJupyter(uid)
       case wsapiWorkflowWebsocket() | apiExecutionsStats() | apiExecutionsResultExport() |
           pveRoute() =>
         checkComputingUnitAccess(uriInfo, headers, bodyOpt)
@@ -77,6 +82,39 @@ object AccessControlResource extends LazyLogging {
     }
   }
 
+  /**
+    * Resolve which JupyterLab pod a request belongs to. This routes; it does not authorize.
+    *
+    * Jupyter is loaded in an iframe and then issues its own requests for assets, contents and
+    * kernel websockets. None of those can carry a Texera token, and there is no session cookie
+    * to fall back on, so the caller cannot be authenticated per request. What protects one
+    * user's notebooks from another is the per-user Jupyter token, which is derived from a
+    * server-held secret and is unguessable; reaching the right pod without it yields a 403 from
+    * Jupyter itself. Cross-pod traffic is blocked separately by a NetworkPolicy.
+    */
+  private def routeToJupyter(uid: String): Response = {
+    val recordedUrl =
+      try {
+        val dao = new UserJupyterDao(SqlServer.getInstance().createDSLContext().configuration())
+        Option(dao.fetchOneByUid(uid.toInt)).map(_.getInternalUrl)
+      } catch {
+        case e: Exception =>
+          logger.error(s"Failed to look up the Jupyter registered for user $uid", e)
+          return Response.status(Response.Status.FORBIDDEN).build()
+      }
+
+    // Envoy routes on an authority, so the scheme and the base path are stripped back off the
+    // address the provisioner recorded.
+    recordedUrl.map(url => new URI(url).getAuthority).filter(a => a != null && a.nonEmpty) match {
+      case Some(authority) =>
+        logger.info(s"Routing Jupyter for user $uid to recorded host: $authority")
+        Response.ok().header("Host", authority).build()
+      case None =>
+        logger.warn(s"Refusing Jupyter for user $uid: no Jupyter is registered")
+        Response.status(Response.Status.FORBIDDEN).build()
+    }
+  }
+
   private def checkComputingUnitAccess(
       uriInfo: UriInfo,
       headers: HttpHeaders,
diff --git a/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala b/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala
index 10ff44db7fd..c13c14b075a 100644
--- a/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala
+++ b/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala
@@ -27,11 +27,13 @@ import org.apache.texera.dao.jooq.generated.enums.{
   WorkflowComputingUnitTypeEnum
 }
 import org.apache.texera.dao.jooq.generated.tables.daos.{
+  UserJupyterDao,
   ComputingUnitUserAccessDao,
   UserDao,
   WorkflowComputingUnitDao
 }
 import org.apache.texera.dao.jooq.generated.tables.pojos.{
+  UserJupyter,
   ComputingUnitUserAccess,
   User,
   WorkflowComputingUnit
@@ -73,6 +75,11 @@ class AccessControlResourceSpec
   private val testNoAccessRecordedUri: String =
     "computing-unit-6.compute-unit-svc.default.svc.cluster.local:7777"
 
+  // What the provisioner records for a user's Jupyter: scheme, authority and the base path
+  // the pod serves under. Only the authority may reach Envoy as a Host header.
+  private val testJupyterInternalUrl: String =
+    "http://jupyter-1.jupyter-svc.texera-jupyter-pool.svc.cluster.local:8888/jupyter/1"
+
   private val testUser1: User = {
     val user = new User()
     user.setUid(1)
@@ -190,6 +197,14 @@ class AccessControlResourceSpec
     readOnlyAccess.setPrivilege(PrivilegeEnum.READ)
     computingUnitOfUserDao.insert(readOnlyAccess)
 
+    // Per-user Jupyter: user 1 has one registered, user 2 deliberately does not.
+    val jupyterDao = new UserJupyterDao(getDSLContext.configuration())
+    val jupyter = new UserJupyter()
+    jupyter.setUid(testUser1.getUid)
+    jupyter.setInternalUrl(testJupyterInternalUrl)
+    jupyter.setPublicUrl("https://texera.example.com/jupyter/1")
+    jupyterDao.insert(jupyter)
+
     token = JwtAuth.jwtToken(JwtAuth.jwtClaims(testUser1))
     token2 = JwtAuth.jwtToken(JwtAuth.jwtClaims(testUser2))
   }
@@ -722,4 +737,56 @@ class AccessControlResourceSpec
     response.getStatus shouldBe Response.Status.OK.getStatusCode
     response.getHeaderString("Host") shouldBe testRecordedUri
   }
+
+  // -- per-user JupyterLab routing --------------------------------------------
+
+  it should "route a Jupyter request to the pod recorded for the uid in the path" in {
+    val (uri, headers) = mockRequest("/jupyter/1/notebooks/work/notebook.ipynb", None)
+    val response = new AccessControlResource().authorizeGet(uri, headers)
+
+    response.getStatus shouldBe Response.Status.OK.getStatusCode
+    // The scheme and base path are stripped: Envoy routes on an authority alone.
+    response.getHeaderString("Host") shouldBe
+      "jupyter-1.jupyter-svc.texera-jupyter-pool.svc.cluster.local:8888"
+  }
+
+  it should "route Jupyter's own subrequests, which carry no token" in {
+    // The iframe's asset and API calls cannot present Texera credentials, so routing has to
+    // work without one. The per-user Jupyter token is what authorizes them.
+    val (uri, headers) =
+      mockRequest("/jupyter/1/api/contents", None, authorizationHeader = None)
+    val response = new AccessControlResource().authorizeGet(uri, headers)
+
+    response.getStatus shouldBe Response.Status.OK.getStatusCode
+    response.getHeaderString("Host") should startWith("jupyter-1.")
+  }
+
+  it should "refuse a Jupyter request for a user with none registered" in {
+    val (uri, headers) = mockRequest("/jupyter/2/tree", None)
+    new AccessControlResource()
+      .authorizeGet(uri, headers)
+      .getStatus shouldBe Response.Status.FORBIDDEN.getStatusCode
+  }
+
+  it should "refuse a Jupyter request for a uid that does not exist" in {
+    val (uri, headers) = mockRequest("/jupyter/999999/tree", None)
+    new AccessControlResource()
+      .authorizeGet(uri, headers)
+      .getStatus shouldBe Response.Status.FORBIDDEN.getStatusCode
+  }
+
+  it should "not treat a Jupyter path without a uid as routable" in {
+    // Falls through to the catch-all, which denies.
+    val (uri, headers) = mockRequest("/jupyter/tree", None)
+    new AccessControlResource()
+      .authorizeGet(uri, headers)
+      .getStatus shouldBe Response.Status.FORBIDDEN.getStatusCode
+  }
+
+  it should "route the gateway-relative form of a Jupyter path" in {
+    val (uri, headers) = mockRequest("auth/jupyter/1/tree", None)
+    new AccessControlResource()
+      .authorizeGet(uri, headers)
+      .getStatus shouldBe Response.Status.OK.getStatusCode
+  }
 }
diff --git a/bin/k8s/templates/base/config-service/config-service-deployment.yaml b/bin/k8s/templates/base/config-service/config-service-deployment.yaml
index f0748785c3a..d6775483eb9 100644
--- a/bin/k8s/templates/base/config-service/config-service-deployment.yaml
+++ b/bin/k8s/templates/base/config-service/config-service-deployment.yaml
@@ -47,6 +47,11 @@ spec:
                 secretKeyRef:
                   name: {{ .Release.Name }}-postgresql
                   key: postgres-password
+            # Shows or hides the notebook migration tool in the workspace. Derived from the
+            # service's own toggle rather than listed in texeraEnvVars, so enabling the tool
+            # is one switch instead of two that can disagree.
+            - name: GUI_WORKFLOW_WORKSPACE_PYTHON_NOTEBOOK_MIGRATION_ENABLED
+              value: "{{ .Values.notebookMigrationService.enabled }}"
             {{- range .Values.texeraEnvVars }}
             - name: {{ .name }}
               value: "{{ .value }}"
diff --git a/bin/k8s/templates/base/gateway/gateway-routes.yaml b/bin/k8s/templates/base/gateway/gateway-routes.yaml
index f07a3157a4e..38c38fb11ee 100644
--- a/bin/k8s/templates/base/gateway/gateway-routes.yaml
+++ b/bin/k8s/templates/base/gateway/gateway-routes.yaml
@@ -74,6 +74,15 @@ spec:
       backendRefs:
         - name: config-service-svc
           port: 9094
+    {{- if .Values.notebookMigrationService.enabled }}
+    - matches:
+        - path:
+            type: PathPrefix
+            value: /api/notebook-migration
+      backendRefs:
+        - name: {{ .Values.notebookMigrationService.name }}-svc
+          port: {{ .Values.notebookMigrationService.service.port }}
+    {{- end }}
     - matches:
         - path:
             type: PathPrefix
@@ -81,6 +90,10 @@ spec:
         - path:
             type: PathPrefix
             value: /api/chat
+      {{- if and .Values.gatewayConfig .Values.gatewayConfig.llmRequestTimeout }}
+      timeouts:
+        request: {{ .Values.gatewayConfig.llmRequestTimeout | quote }}
+      {{- end }}
       backendRefs:
         - name: access-control-service-svc
           port: 9096
@@ -135,6 +148,13 @@ spec:
         - path:
             type: PathPrefix
             value: /api/pve
+        {{- if .Values.notebookMigrationService.enabled }}
+        # Per-user JupyterLab. ExtAuthz reads the uid from the path and rewrites Host to
+        # that user's pod; the per-user Jupyter token is what authorizes the request.
+        - path:
+            type: PathPrefix
+            value: /jupyter
+        {{- end }}
       backendRefs:
         - group: gateway.envoyproxy.io
           kind: Backend
diff --git a/bin/k8s/templates/base/jupyter-pool/jupyter-namespace.yaml b/bin/k8s/templates/base/jupyter-pool/jupyter-namespace.yaml
new file mode 100644
index 00000000000..9e228d919e7
--- /dev/null
+++ b/bin/k8s/templates/base/jupyter-pool/jupyter-namespace.yaml
@@ -0,0 +1,22 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+{{- if and .Values.notebookMigrationService.enabled .Values.jupyterPool.createNamespaces }}
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: {{ .Values.jupyterPool.namespace }}
+{{- end }}
diff --git a/bin/k8s/templates/base/jupyter-pool/jupyter-network-policy.yaml b/bin/k8s/templates/base/jupyter-pool/jupyter-network-policy.yaml
new file mode 100644
index 00000000000..575b92b264a
--- /dev/null
+++ b/bin/k8s/templates/base/jupyter-pool/jupyter-network-policy.yaml
@@ -0,0 +1,48 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+{{- if and .Values.notebookMigrationService.enabled .Values.jupyterPool.networkPolicy.enabled }}
+# Stops one user's JupyterLab from reaching another's. Users run arbitrary code in these
+# pods, so a neighbour in the pool is the one genuinely hostile caller. Allowing every
+# namespace but the pool's own denies pod-to-pod traffic inside it while leaving the real
+# callers working: the notebook migration service, and the Envoy proxy wherever the gateway
+# installation runs it.
+#
+# Defence in depth, not the authorisation boundary: the per-user Jupyter token is what stops
+# one user reading another's notebooks. Egress is left alone, since notebooks legitimately
+# install packages and call out.
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: {{ .Values.jupyterPool.name }}-deny-cross-user
+  namespace: {{ .Values.jupyterPool.namespace }}
+spec:
+  podSelector:
+    matchLabels:
+      type: jupyter
+  policyTypes:
+    - Ingress
+  ingress:
+    - from:
+        # kubernetes.io/metadata.name is set automatically on every namespace, so this
+        # selects "any namespace but the pool's own" without labelling anything by hand.
+        - namespaceSelector:
+            matchExpressions:
+              - key: kubernetes.io/metadata.name
+                operator: NotIn
+                values:
+                  - {{ .Values.jupyterPool.namespace }}
+{{- end }}
diff --git a/bin/k8s/templates/base/jupyter-pool/jupyter-prepull-daemonset.yaml b/bin/k8s/templates/base/jupyter-pool/jupyter-prepull-daemonset.yaml
new file mode 100644
index 00000000000..e6fca48a276
--- /dev/null
+++ b/bin/k8s/templates/base/jupyter-pool/jupyter-prepull-daemonset.yaml
@@ -0,0 +1,57 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+{{- if and .Values.notebookMigrationService.enabled .Values.jupyterPool.prepullImage }}
+# Pulls the JupyterLab image onto every node ahead of time. Pods are created on demand and
+# the service waits a bounded time for one to answer, so a first-time pull on a cold node can
+# outlast that wait and the provisioning attempt is discarded. Mirrors the computing unit
+# pool's prepuller. Set jupyterPool.prepullImage to false to trade cold starts for one fewer
+# pod per node.
+apiVersion: apps/v1
+kind: DaemonSet
+metadata:
+  name: {{ .Release.Name }}-jupyter-prepuller
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ .Release.Name }}-jupyter-prepuller
+spec:
+  selector:
+    matchLabels:
+      app: {{ .Release.Name }}-jupyter-prepuller
+  template:
+    metadata:
+      labels:
+        app: {{ .Release.Name }}-jupyter-prepuller
+    spec:
+      restartPolicy: Always
+      tolerations:
+        - operator: "Exists"
+      initContainers:
+        - name: prepuller
+          image: {{ .Values.texera.imageRegistry }}/{{ .Values.jupyterPool.imageName }}:{{ .Values.texera.imageTag }}
+          imagePullPolicy: {{ .Values.texeraImages.pullPolicy }}
+          command: ["sh", "-c", "true"]
+      containers:
+        - name: pause
+          image: gcr.io/google_containers/pause:3.2
+          resources:
+            limits:
+              cpu: 1m
+              memory: 8Mi
+            requests:
+              cpu: 1m
+              memory: 8Mi
+{{- end }}
diff --git a/bin/k8s/templates/base/jupyter-pool/jupyter-resource-quota.yaml b/bin/k8s/templates/base/jupyter-pool/jupyter-resource-quota.yaml
new file mode 100644
index 00000000000..7e186b18eb6
--- /dev/null
+++ b/bin/k8s/templates/base/jupyter-pool/jupyter-resource-quota.yaml
@@ -0,0 +1,29 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+{{- if and .Values.notebookMigrationService.enabled .Values.jupyterPool.createNamespaces .Values.jupyterPool.maxRequestedResources }}
+# Ceiling for the pool as a whole. Pods are created on demand, one per user, so without this
+# a busy deployment has no upper bound on what the tool can consume.
+apiVersion: v1
+kind: ResourceQuota
+metadata:
+  name: {{ .Values.jupyterPool.name }}-resource-quota
+  namespace: {{ .Values.jupyterPool.namespace }}
+spec:
+  hard:
+    requests.cpu: "{{ .Values.jupyterPool.maxRequestedResources.cpu }}"
+    requests.memory: {{ .Values.jupyterPool.maxRequestedResources.memory }}
+{{- end }}
diff --git a/bin/k8s/templates/base/jupyter-pool/jupyter-service.yaml b/bin/k8s/templates/base/jupyter-pool/jupyter-service.yaml
new file mode 100644
index 00000000000..2f0fb1f5e92
--- /dev/null
+++ b/bin/k8s/templates/base/jupyter-pool/jupyter-service.yaml
@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+{{- if .Values.notebookMigrationService.enabled }}
+# Headless, so each user's pod is addressable individually rather than load balanced across
+# the pool: <pod>.<this service>.<namespace>.svc.cluster.local. The notebook migration
+# service creates pods whose hostname is the pod name and whose subdomain is this service's
+# name, which is what makes that address resolve. The selector matches the "type" label the
+# service stamps on every JupyterLab pod it creates.
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ .Values.jupyterPool.name }}-svc
+  namespace: {{ .Values.jupyterPool.namespace }}
+spec:
+  clusterIP: None
+  selector:
+    type: jupyter
+  ports:
+    - protocol: TCP
+      port: {{ .Values.jupyterPool.service.port }}
+      targetPort: {{ .Values.jupyterPool.service.targetPort }}
+{{- end }}
diff --git a/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-deployment.yaml b/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-deployment.yaml
new file mode 100644
index 00000000000..1257e756c31
--- /dev/null
+++ b/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-deployment.yaml
@@ -0,0 +1,99 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+{{- if .Values.notebookMigrationService.enabled }}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ .Release.Name }}-{{ .Values.notebookMigrationService.name }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ .Release.Name }}-{{ .Values.notebookMigrationService.name }}
+spec:
+  replicas: {{ .Values.notebookMigrationService.numOfPods | default 1 }}
+  selector:
+    matchLabels:
+      app: {{ .Release.Name }}-{{ .Values.notebookMigrationService.name }}
+  template:
+    metadata:
+      labels:
+        app: {{ .Release.Name }}-{{ .Values.notebookMigrationService.name }}
+    spec:
+      # Needed to create and delete each user's JupyterLab pod in the pool namespace.
+      serviceAccountName: {{ .Values.notebookMigrationService.serviceAccountName }}
+      containers:
+        - name: {{ .Values.notebookMigrationService.name }}
+          image: {{ .Values.texera.imageRegistry }}/{{ .Values.notebookMigrationService.imageName }}:{{ .Values.texera.imageTag }}
+          imagePullPolicy: {{ .Values.texeraImages.pullPolicy }}
+          ports:
+            - containerPort: {{ .Values.notebookMigrationService.service.port }}
+          env:
+            - name: STORAGE_JDBC_URL
+              value: jdbc:postgresql://{{ .Release.Name }}-postgresql:5432/texera_db?currentSchema=texera_db,public
+            - name: STORAGE_JDBC_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: {{ .Release.Name }}-postgresql
+                  key: postgres-password
+            # Resolve each user's JupyterLab rather than one shared server.
+            - name: KUBERNETES_JUPYTER_ENABLED
+              value: "true"
+            - name: KUBERNETES_JUPYTER_NAMESPACE
+              value: {{ .Values.jupyterPool.namespace }}
+            - name: KUBERNETES_JUPYTER_SERVICE_NAME
+              value: {{ .Values.jupyterPool.name }}-svc
+            - name: KUBERNETES_JUPYTER_IMAGE_NAME
+              value: {{ .Values.texera.imageRegistry }}/{{ .Values.jupyterPool.imageName }}:{{ .Values.texera.imageTag }}
+            - name: KUBERNETES_JUPYTER_CPU_LIMIT
+              value: "{{ .Values.jupyterPool.resources.cpuLimit }}"
+            - name: KUBERNETES_JUPYTER_MEMORY_LIMIT
+              value: {{ .Values.jupyterPool.resources.memoryLimit }}
+            # The pod's own prefix and the browser-facing address are rendered from one
+            # basePath, so they cannot drift apart.
+            - name: KUBERNETES_JUPYTER_BASE_URL
+              value: {{ .Values.jupyterPool.basePath }}
+            {{- $origin := .Values.notebookMigrationService.publicOrigin }}
+            {{- if and (not $origin) .Values.gatewayConfig .Values.gatewayConfig.hostname }}
+            {{- $origin = printf "https://%s" .Values.gatewayConfig.hostname }}
+            {{- end }}
+            {{- if $origin }}
+            - name: KUBERNETES_JUPYTER_PUBLIC_URL_TEMPLATE
+              value: {{ $origin }}{{ .Values.jupyterPool.basePath }}/{uid}
+            - name: KUBERNETES_JUPYTER_TEXERA_ORIGIN
+              value: {{ $origin }}
+            {{- end }}
+            - name: JUPYTER_TOKEN_SECRET
+              valueFrom:
+                secretKeyRef:
+                  name: {{ .Release.Name }}-notebook-migration-service-secret
+                  key: jupyter-token-secret
+            {{- range .Values.texeraEnvVars }}
+            - name: {{ .name }}
+              value: "{{ .value }}"
+            {{- end }}
+          livenessProbe:
+            httpGet:
+              path: /api/healthcheck
+              port: {{ .Values.notebookMigrationService.service.port }}
+            initialDelaySeconds: 30
+            periodSeconds: 10
+          readinessProbe:
+            httpGet:
+              path: /api/healthcheck
+              port: {{ .Values.notebookMigrationService.service.port }}
+            initialDelaySeconds: 5
+            periodSeconds: 5
+{{- end }}
diff --git a/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-secret.yaml b/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-secret.yaml
new file mode 100644
index 00000000000..c5fb3a92aee
--- /dev/null
+++ b/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-secret.yaml
@@ -0,0 +1,29 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+{{- if .Values.notebookMigrationService.enabled }}
+# Key the per-user JupyterLab tokens are derived from. Kept in a Secret rather than the
+# deployment's env list because it is a credential: anyone holding it can derive any user's
+# token.
+apiVersion: v1
+kind: Secret
+metadata:
+  name: {{ .Release.Name }}-notebook-migration-service-secret
+  namespace: {{ .Release.Namespace }}
+type: Opaque
+stringData:
+  jupyter-token-secret: "{{ .Values.notebookMigrationService.jupyterTokenSecret }}"
+{{- end }}
diff --git a/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-service-account.yaml b/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-service-account.yaml
new file mode 100644
index 00000000000..1b5d37ec3a6
--- /dev/null
+++ b/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-service-account.yaml
@@ -0,0 +1,49 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+{{- if .Values.notebookMigrationService.enabled }}
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: {{ .Values.notebookMigrationService.serviceAccountName }}
+  namespace: {{ .Release.Namespace }}
+---
+# Scoped to the JupyterLab pool namespace only: the service starts and stops a user's own
+# notebook server and needs nothing in the release namespace.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: {{ .Values.notebookMigrationService.name }}
+  namespace: {{ .Values.jupyterPool.namespace }}
+rules:
+  - apiGroups: [""]
+    resources: ["pods"]
+    verbs: ["get", "list", "watch", "create", "delete"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: {{ .Values.notebookMigrationService.name }}-binding
+  namespace: {{ .Values.jupyterPool.namespace }}
+subjects:
+  - kind: ServiceAccount
+    name: {{ .Values.notebookMigrationService.serviceAccountName }}
+    namespace: {{ .Release.Namespace }}
+roleRef:
+  kind: Role
+  name: {{ .Values.notebookMigrationService.name }}
+  apiGroup: rbac.authorization.k8s.io
+{{- end }}
diff --git a/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-service.yaml b/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-service.yaml
new file mode 100644
index 00000000000..d13a2d47990
--- /dev/null
+++ b/bin/k8s/templates/base/notebook-migration-service/notebook-migration-service-service.yaml
@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+{{- if .Values.notebookMigrationService.enabled }}
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ .Values.notebookMigrationService.name }}-svc
+  namespace: {{ .Release.Namespace }}
+spec:
+  type: {{ .Values.notebookMigrationService.service.type }}
+  selector:
+    app: {{ .Release.Name }}-{{ .Values.notebookMigrationService.name }}
+  ports:
+    - protocol: TCP
+      port: {{ .Values.notebookMigrationService.service.port }}
+      targetPort: {{ .Values.notebookMigrationService.service.port }}
+{{- end }}
diff --git a/bin/k8s/values-development.yaml b/bin/k8s/values-development.yaml
index f2f885f105e..b01d5e738e7 100644
--- a/bin/k8s/values-development.yaml
+++ b/bin/k8s/values-development.yaml
@@ -295,6 +295,27 @@ workflowComputingUnitPool:
     port: 8085
     targetPort: 8085
 
+# Notebook migration tool. The base chart ships it off; development turns it on.
+notebookMigrationService:
+  enabled: true
+  # Reached through a port-forward, so there is no DNS name to derive an origin from.
+  publicOrigin: "http://localhost:30080"
+
+# Per-user JupyterLab pods, sized for a single-node development cluster. Only the values that
+# differ from the base chart are listed, since Helm merges the rest.
+jupyterPool:
+  # The base default of a full CPU and 2Gi per user exhausts a laptop cluster after two of
+  # them. Limits become requests here, because the pods declare no requests of their own.
+  resources:
+    cpuLimit: "0.5"
+    memoryLimit: 1Gi
+  # Low enough to actually bind, so a provisioning loop cannot fill the node.
+  maxRequestedResources:
+    cpu: 4
+    memory: 8Gi
+  # One node, so prepulling saves nothing and costs a large image pull at install time.
+  prepullImage: false
+
 texeraEnvVars:
   - name: USER_SYS_ADMIN_USERNAME
     value: "texera"
diff --git a/bin/k8s/values.yaml b/bin/k8s/values.yaml
index 907c212cb6b..23599bf083a 100644
--- a/bin/k8s/values.yaml
+++ b/bin/k8s/values.yaml
@@ -195,6 +195,61 @@ webserver:
     type: ClusterIP
     port: 8080
 
+notebookMigrationService:
+  # Turns the whole notebook migration tool on or off: the service, its route, the per-user
+  # JupyterLab pool, and the button in the workspace.
+  enabled: false
+  name: notebook-migration-service
+  numOfPods: 1
+  serviceAccountName: notebook-migration-service-service-account
+  imageName: texera-notebook-migration-service
+  service:
+    type: ClusterIP
+    port: 9098
+  # Origin the browser reaches Texera on, used for the JupyterLab iframe URL and for the
+  # CSP that lets Texera embed it. Required wherever there is no DNS name, such as a
+  # port-forward or a NodePort. Falls back to the gateway hostname when left empty.
+  publicOrigin: ""
+  # HMAC key each user's JupyterLab token is derived from. Nothing is stored, so this must
+  # stay stable across restarts or previously issued tokens stop matching.
+  # Development-only default. Production environments MUST override this with a different,
+  # securely generated secret.
+  jupyterTokenSecret: "c4e1f7a9b2d5c8e0f3a6b9d2e5f8a1c4b7d0e3f6a9c2b5d8e1f4a7c0b3d6e9f2"
+
+# Per-user JupyterLab pods, the stateful half of the notebook migration tool. One pod per
+# user, addressed through the headless service below.
+jupyterPool:
+  createNamespaces: true
+  name: texera-jupyter
+  # Note: like the computing unit pool, this namespace can collide when several Texera
+  # deployments share a cluster.
+  namespace: texera-jupyter-pool
+  imageName: texera-jupyter
+  # Pull the image onto every node ahead of time, so a first-time pull cannot outlast the
+  # bounded wait for a new pod to answer. Costs one small pod per node.
+  prepullImage: true
+  # Path prefix each user's JupyterLab is served under; the uid is appended, so user 7 is
+  # served at /jupyter/7/. The gateway reads that uid to pick the pod.
+  basePath: /jupyter
+  # Must match kubernetes.jupyter-port-num, which the service reads from its own config
+  # and no environment variable overrides.
+  service:
+    port: 8888
+    targetPort: 8888
+  # Per-pod limits.
+  resources:
+    cpuLimit: "1"
+    memoryLimit: 2Gi
+  networkPolicy:
+    # Denies pod-to-pod traffic inside the pool, so one user's JupyterLab cannot reach
+    # another's. Requires a cluster with a NetworkPolicy controller; without one the object
+    # is created but not enforced.
+    enabled: true
+  # Ceiling for the pool as a whole.
+  maxRequestedResources:
+    cpu: 50
+    memory: 50Gi
+
 workflowComputingUnitManager:
   name: workflow-computing-unit-manager
   numOfPods: 1
@@ -396,6 +451,13 @@ metrics-server:
 gatewayConfig:
   # Routes are available at bin/k8s/templates/gateway-routes.yaml
 
+  # Ceiling for LLM requests through /api/chat and /api/models. Completions routinely run
+  # for tens of seconds, and Envoy's default route timeout is 15s, which severs them while
+  # the upstream call is still in flight and then succeeds unseen. Keep this at or above
+  # GUI_WORKFLOW_WORKSPACE_PYTHON_NOTEBOOK_MIGRATION_TIMEOUT_MINUTES, which is what the
+  # frontend is prepared to wait.
+  llmRequestTimeout: 10m
+
   # The hostname for the Gateway listener (HTTP/HTTPS).
   # e.g., "texera.example.com"
   hostname: ""
diff --git a/common/config/src/main/resources/kubernetes.conf b/common/config/src/main/resources/kubernetes.conf
index e85924e570c..1c19e89c783 100644
--- a/common/config/src/main/resources/kubernetes.conf
+++ b/common/config/src/main/resources/kubernetes.conf
@@ -54,4 +54,44 @@ kubernetes {
   # GPU resource key used in Kubernetes (vendor-specific)
   computing-unit-gpu-resource-key = "nvidia.com/gpu"
   computing-unit-gpu-resource-key = ${?KUBERNETES_COMPUTING_UNIT_GPU_RESOURCE_KEY}
+
+  # Per-user JupyterLab pods. Separate from `enabled` above, so a deployment can run
+  # computing units on Kubernetes without per-user Jupyter. While this is off, the
+  # notebook migration service uses the single Jupyter from storage.jupyter.
+  jupyter-enabled = false
+  jupyter-enabled = ${?KUBERNETES_JUPYTER_ENABLED}
+
+  jupyter-namespace = "texera-jupyter-pool"
+  jupyter-namespace = ${?KUBERNETES_JUPYTER_NAMESPACE}
+
+  jupyter-service-name = "jupyter-svc"
+  jupyter-service-name = ${?KUBERNETES_JUPYTER_SERVICE_NAME}
+
+  jupyter-image-name = "ghcr.io/apache/texera-jupyter:latest"
+  jupyter-image-name = ${?KUBERNETES_JUPYTER_IMAGE_NAME}
+
+  jupyter-port-num = 8888
+
+  # Prefix each user's Jupyter serves under; the uid is appended, so user 7 is served at
+  # <prefix>/7/. The gateway reads that uid to pick the pod, so this has to match the path
+  # component of jupyter-public-url-template.
+  jupyter-base-url = "/jupyter"
+  jupyter-base-url = ${?KUBERNETES_JUPYTER_BASE_URL}
+
+  # Origin the Jupyter pod names in its iframe CSP and postMessage checks. Empty leaves the
+  # image's own default, which only suits local development.
+  jupyter-texera-origin = ""
+  jupyter-texera-origin = ${?KUBERNETES_JUPYTER_TEXERA_ORIGIN}
+
+  jupyter-cpu-limit = "1"
+  jupyter-cpu-limit = ${?KUBERNETES_JUPYTER_CPU_LIMIT}
+
+  jupyter-memory-limit = "2Gi"
+  jupyter-memory-limit = ${?KUBERNETES_JUPYTER_MEMORY_LIMIT}
+
+  # Browser-facing address, with {uid} substituted. The in-network pod name does not
+  # resolve from the browser, so a deployment that publishes Jupyter sets this; empty
+  # falls back to the in-network address.
+  jupyter-public-url-template = ""
+  jupyter-public-url-template = ${?KUBERNETES_JUPYTER_PUBLIC_URL_TEMPLATE}
 }
\ No newline at end of file
diff --git a/common/config/src/main/resources/storage.conf b/common/config/src/main/resources/storage.conf
index 9af2924901d..b6b76d8ad39 100644
--- a/common/config/src/main/resources/storage.conf
+++ b/common/config/src/main/resources/storage.conf
@@ -177,7 +177,8 @@ storage {
         password = ${?STORAGE_JDBC_PASSWORD}
     }
 
-    # Configurations of the JupyterLab service
+    # The single JupyterLab used when per-user provisioning (kubernetes.jupyter-enabled)
+    # is off, which is how the single-node and local-dev deployments run.
     jupyter {
         internal-url = "http://localhost:9100"
         internal-url = ${?STORAGE_JUPYTER_INTERNAL_URL}
@@ -188,5 +189,11 @@ storage {
         # Read from the same JUPYTER_TOKEN env var as the Jupyter container
         token = "texera"
         token = ${?JUPYTER_TOKEN}
+
+        # HMAC key each per-user Jupyter token is derived from, so no token is stored.
+        # Required when per-user Jupyter is on, and must stay stable across restarts and
+        # replicas or previously issued tokens stop matching.
+        token-secret = ""
+        token-secret = ${?JUPYTER_TOKEN_SECRET}
     }
 }
diff --git a/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala
index f6294767365..9766792b37e 100644
--- a/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala
+++ b/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala
@@ -64,4 +64,19 @@ object KubernetesConfig {
 
   // GPU resource key used directly in Kubernetes resource specifications
   val gpuResourceKey: String = conf.getString("kubernetes.computing-unit-gpu-resource-key")
+
+  // Per-user JupyterLab pods, gated independently of computing units.
+  val jupyterEnabled: Boolean = conf.getBoolean("kubernetes.jupyter-enabled")
+  val jupyterNamespace: String = conf.getString("kubernetes.jupyter-namespace")
+  val jupyterServiceName: String = conf.getString("kubernetes.jupyter-service-name")
+  val jupyterImageName: String = conf.getString("kubernetes.jupyter-image-name")
+  val jupyterPortNumber: Int = conf.getInt("kubernetes.jupyter-port-num")
+  val jupyterBaseUrl: String = conf.getString("kubernetes.jupyter-base-url")
+  val jupyterTexeraOrigin: String = conf.getString("kubernetes.jupyter-texera-origin")
+  val jupyterCpuLimit: String = conf.getString("kubernetes.jupyter-cpu-limit")
+  val jupyterMemoryLimit: String = conf.getString("kubernetes.jupyter-memory-limit")
+
+  // Browser-facing address with {uid} substituted; empty means use the in-network one.
+  val jupyterPublicUrlTemplate: String =
+    conf.getString("kubernetes.jupyter-public-url-template")
 }
diff --git a/common/config/src/main/scala/org/apache/texera/common/config/StorageConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/StorageConfig.scala
index e48fe4f84ee..2627b12f5d2 100644
--- a/common/config/src/main/scala/org/apache/texera/common/config/StorageConfig.scala
+++ b/common/config/src/main/scala/org/apache/texera/common/config/StorageConfig.scala
@@ -160,4 +160,7 @@ object StorageConfig {
   val jupyterInternalURL: String = conf.getString("storage.jupyter.internal-url")
   val jupyterPublicURL: String = conf.getString("storage.jupyter.public-url")
   val jupyterToken: String = conf.getString("storage.jupyter.token")
+
+  // HMAC key for per-user token derivation; empty unless a deployment sets it.
+  val jupyterTokenSecret: String = conf.getString("storage.jupyter.token-secret")
 }
diff --git a/common/config/src/test/scala/org/apache/texera/common/config/KubernetesConfigSpec.scala b/common/config/src/test/scala/org/apache/texera/common/config/KubernetesConfigSpec.scala
index ba9a0acd7f0..0b8cb9ceb33 100644
--- a/common/config/src/test/scala/org/apache/texera/common/config/KubernetesConfigSpec.scala
+++ b/common/config/src/test/scala/org/apache/texera/common/config/KubernetesConfigSpec.scala
@@ -24,7 +24,7 @@ import org.scalatest.matchers.should.Matchers
 
 /**
   * Spec for [[KubernetesConfig]]. Reading each value forces resolution from kubernetes.conf, so a
-  * renamed or mistyped key surfaces here as a ConfigException. Every value except the port number
+  * renamed or mistyped key surfaces here as a ConfigException. Every value except the port numbers
   * carries a `${?ENV}` override, so exact-value assertions are guarded on the env var being unset.
   */
 class KubernetesConfigSpec extends AnyFlatSpec with Matchers {
@@ -70,6 +70,35 @@ class KubernetesConfigSpec extends AnyFlatSpec with Matchers {
     KubernetesConfig.maxNumOfRunningComputingUnitsPerUser should be >= 0
   }
 
+  "KubernetesConfig jupyter settings" should "resolve to their kubernetes.conf defaults" in {
+    KubernetesConfig.jupyterPortNumber shouldBe 8888
+    // Off by default and keyed separately from kubernetes.enabled, so enabling computing
+    // units on Kubernetes never silently enables per-user Jupyter.
+    ifUnset("KUBERNETES_JUPYTER_ENABLED")(KubernetesConfig.jupyterEnabled shouldBe false)
+    ifUnset("KUBERNETES_JUPYTER_NAMESPACE")(
+      KubernetesConfig.jupyterNamespace shouldBe "texera-jupyter-pool"
+    )
+    ifUnset("KUBERNETES_JUPYTER_SERVICE_NAME")(
+      KubernetesConfig.jupyterServiceName shouldBe "jupyter-svc"
+    )
+    ifUnset("KUBERNETES_JUPYTER_IMAGE_NAME")(
+      KubernetesConfig.jupyterImageName shouldBe "ghcr.io/apache/texera-jupyter:latest"
+    )
+    // A prefix, not a full path: the provisioner appends the uid.
+    ifUnset("KUBERNETES_JUPYTER_BASE_URL")(KubernetesConfig.jupyterBaseUrl shouldBe "/jupyter")
+    // Empty by default: only a real deployment knows its own origin.
+    ifUnset("KUBERNETES_JUPYTER_TEXERA_ORIGIN")(KubernetesConfig.jupyterTexeraOrigin shouldBe "")
+    ifUnset("KUBERNETES_JUPYTER_CPU_LIMIT")(KubernetesConfig.jupyterCpuLimit shouldBe "1")
+    ifUnset("KUBERNETES_JUPYTER_MEMORY_LIMIT")(
+      KubernetesConfig.jupyterMemoryLimit shouldBe "2Gi"
+    )
+    // Empty means the browser is handed the in-network address; a deployment that
+    // publishes Jupyter overrides it.
+    ifUnset("KUBERNETES_JUPYTER_PUBLIC_URL_TEMPLATE")(
+      KubernetesConfig.jupyterPublicUrlTemplate shouldBe ""
+    )
+  }
+
   "KubernetesConfig limit options" should "parse into trimmed, non-empty lists" in {
     ifUnset("KUBERNETES_COMPUTING_UNIT_CPU_LIMIT_OPTIONS")(
       KubernetesConfig.cpuLimitOptions shouldBe List("1", "2", "4")
diff --git a/common/config/src/test/scala/org/apache/texera/common/config/StorageConfigSpec.scala b/common/config/src/test/scala/org/apache/texera/common/config/StorageConfigSpec.scala
index ac34c467646..45b628a3190 100644
--- a/common/config/src/test/scala/org/apache/texera/common/config/StorageConfigSpec.scala
+++ b/common/config/src/test/scala/org/apache/texera/common/config/StorageConfigSpec.scala
@@ -73,4 +73,12 @@ class StorageConfigSpec extends AnyFlatSpec with Matchers {
       StorageConfig.jupyterPublicURL shouldBe StorageConfig.jupyterInternalURL
     }
   }
+
+  it should "default the token secret to empty so a deployment must set it deliberately" in {
+    // Per-user tokens are derived from this key, so it has no safe default: an empty
+    // value must be caught at start-up rather than silently deriving from nothing.
+    if (sys.env.get("JUPYTER_TOKEN_SECRET").isEmpty) {
+      StorageConfig.jupyterTokenSecret shouldBe ""
+    }
+  }
 }
diff --git a/notebook-migration-service/LICENSE-binary b/notebook-migration-service/LICENSE-binary
index 78f1df46a94..7b9b8f84b45 100644
--- a/notebook-migration-service/LICENSE-binary
+++ b/notebook-migration-service/LICENSE-binary
@@ -224,10 +224,10 @@ Scala/Java jars:
   - com.fasterxml.jackson.core.jackson-annotations-2.18.8.jar
   - com.fasterxml.jackson.core.jackson-core-2.18.8.jar
   - com.fasterxml.jackson.core.jackson-databind-2.18.8.jar
-  - com.fasterxml.jackson.dataformat.jackson-dataformat-yaml-2.16.1.jar
+  - com.fasterxml.jackson.dataformat.jackson-dataformat-yaml-2.17.0.jar
   - com.fasterxml.jackson.datatype.jackson-datatype-guava-2.16.1.jar
   - com.fasterxml.jackson.datatype.jackson-datatype-jdk8-2.16.1.jar
-  - com.fasterxml.jackson.datatype.jackson-datatype-jsr310-2.16.1.jar
+  - com.fasterxml.jackson.datatype.jackson-datatype-jsr310-2.17.0.jar
   - com.fasterxml.jackson.jakarta.rs.jackson-jakarta-rs-base-2.16.1.jar
   - com.fasterxml.jackson.jakarta.rs.jackson-jakarta-rs-json-provider-2.16.1.jar
   - com.fasterxml.jackson.module.jackson-module-blackbird-2.16.1.jar
@@ -242,6 +242,9 @@ Scala/Java jars:
   - com.google.guava.listenablefuture-9999.0-empty-to-avoid-conflict-with-guava.jar
   - com.google.j2objc.j2objc-annotations-2.8.jar
   - com.helger.profiler-1.1.1.jar
+  - com.squareup.okhttp3.logging-interceptor-3.12.12.jar
+  - com.squareup.okhttp3.okhttp-3.12.12.jar
+  - com.squareup.okio.okio-1.15.0.jar
   - com.thesamet.scalapb.lenses_2.13-0.11.20.jar
   - com.thesamet.scalapb.scalapb-json4s_2.13-0.12.0.jar
   - com.thesamet.scalapb.scalapb-runtime_2.13-0.11.20.jar
@@ -274,6 +277,32 @@ Scala/Java jars:
   - io.dropwizard.metrics.metrics-json-4.2.25.jar
   - io.dropwizard.metrics.metrics-jvm-4.2.25.jar
   - io.dropwizard.metrics.metrics-logback-4.2.25.jar
+  - io.fabric8.kubernetes-client-6.12.1.jar
+  - io.fabric8.kubernetes-client-api-6.12.1.jar
+  - io.fabric8.kubernetes-httpclient-okhttp-6.12.1.jar
+  - io.fabric8.kubernetes-model-admissionregistration-6.12.1.jar
+  - io.fabric8.kubernetes-model-apiextensions-6.12.1.jar
+  - io.fabric8.kubernetes-model-apps-6.12.1.jar
+  - io.fabric8.kubernetes-model-autoscaling-6.12.1.jar
+  - io.fabric8.kubernetes-model-batch-6.12.1.jar
+  - io.fabric8.kubernetes-model-certificates-6.12.1.jar
+  - io.fabric8.kubernetes-model-common-6.12.1.jar
+  - io.fabric8.kubernetes-model-coordination-6.12.1.jar
+  - io.fabric8.kubernetes-model-core-6.12.1.jar
+  - io.fabric8.kubernetes-model-discovery-6.12.1.jar
+  - io.fabric8.kubernetes-model-events-6.12.1.jar
+  - io.fabric8.kubernetes-model-extensions-6.12.1.jar
+  - io.fabric8.kubernetes-model-flowcontrol-6.12.1.jar
+  - io.fabric8.kubernetes-model-gatewayapi-6.12.1.jar
+  - io.fabric8.kubernetes-model-metrics-6.12.1.jar
+  - io.fabric8.kubernetes-model-networking-6.12.1.jar
+  - io.fabric8.kubernetes-model-node-6.12.1.jar
+  - io.fabric8.kubernetes-model-policy-6.12.1.jar
+  - io.fabric8.kubernetes-model-rbac-6.12.1.jar
+  - io.fabric8.kubernetes-model-resource-6.12.1.jar
+  - io.fabric8.kubernetes-model-scheduling-6.12.1.jar
+  - io.fabric8.kubernetes-model-storageclass-6.12.1.jar
+  - io.fabric8.zjsonpatch-0.3.0.jar
   - io.r2dbc.r2dbc-spi-1.0.0.RELEASE.jar
   - jakarta.inject.jakarta.inject-api-2.0.1.jar
   - jakarta.validation.jakarta.validation-api-3.0.2.jar
@@ -300,6 +329,7 @@ Scala/Java jars:
   - org.scala-lang.scala-reflect-2.13.18.jar
   - org.slf4j.jcl-over-slf4j-2.0.12.jar
   - org.slf4j.log4j-over-slf4j-2.0.12.jar
+  - org.snakeyaml.snakeyaml-engine-2.7.jar
   - org.yaml.snakeyaml-2.2.jar
 
 --------------------------------------------------------------------------------
@@ -327,7 +357,7 @@ Scala/Java jars:
   - net.sourceforge.argparse4j.argparse4j-0.9.0.jar
   - org.checkerframework.checker-qual-3.52.0.jar
   - org.slf4j.jul-to-slf4j-2.0.12.jar
-  - org.slf4j.slf4j-api-2.0.12.jar
+  - org.slf4j.slf4j-api-2.0.13.jar
 
 --------------------------------------------------------------------------------
 Dependencies under the BSD 3-Clause License
diff --git a/notebook-migration-service/build.sbt b/notebook-migration-service/build.sbt
index 53dc3c9e315..84b48e24fdc 100644
--- a/notebook-migration-service/build.sbt
+++ b/notebook-migration-service/build.sbt
@@ -83,5 +83,6 @@ libraryDependencies ++= Seq(
 libraryDependencies ++= Seq(
   "io.dropwizard" % "dropwizard-core" % dropwizardVersion,
   "io.dropwizard" % "dropwizard-auth" % dropwizardVersion, // Dropwizard Authentication module
-  "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.18.8"
+  "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.18.8",
+  "io.fabric8" % "kubernetes-client" % "6.12.1" // Provisions per-user JupyterLab pods
 )
\ No newline at end of file
diff --git a/notebook-migration-service/src/main/resources/start-texera-jupyter.sh b/notebook-migration-service/src/main/resources/start-texera-jupyter.sh
index 2bfb5a3baff..b38be026040 100644
--- a/notebook-migration-service/src/main/resources/start-texera-jupyter.sh
+++ b/notebook-migration-service/src/main/resources/start-texera-jupyter.sh
@@ -19,14 +19,22 @@
 set -euo pipefail
 
 # Texera app origin used by custom.js (postMessage targetOrigin + inbound origin
-# check) and by the iframe CSP frame-ancestors. Override TEXERA_ORIGIN for
-# deployments under a real hostname; defaults to the local dev origin.
+# check), by the iframe CSP frame-ancestors, and by Jupyter's own cross-origin check.
+# That last one matters wherever a proxy rewrites the Host header: Jupyter compares
+# Origin against Host and rejects cookie-authenticated API calls when they differ, which
+# leaves the notebook without a kernel. Override TEXERA_ORIGIN for deployments under a
+# real hostname; defaults to the local dev origin.
 TEXERA_ORIGIN="${TEXERA_ORIGIN:-http://localhost:4200}"
 
 # Weak default token so the server is not fully open to anyone reachable on the
 # published port. The Texera-side iframe URL must pass this through ?token=<value>.
 JUPYTER_TOKEN="${JUPYTER_TOKEN:-texera}"
 
+# Path Jupyter serves under. A deployment that puts every user's Jupyter on one hostname
+# routes by path, so the server has to know its own prefix. Defaults to "/", which is what
+# single-node and local dev use.
+JUPYTER_BASE_URL="${JUPYTER_BASE_URL:-/}"
+
 # Substitute the origin placeholder in custom.js before the server starts serving it.
 sed -i "s|__TEXERA_ORIGIN__|${TEXERA_ORIGIN}|g" /home/jovyan/.jupyter/custom/custom.js
 
@@ -34,5 +42,7 @@ exec start-notebook.sh \
   --NotebookApp.token="${JUPYTER_TOKEN}" \
   --NotebookApp.password='' \
   --NotebookApp.disable_check_xsrf=True \
+  --NotebookApp.allow_origin="${TEXERA_ORIGIN}" \
   --NotebookApp.tornado_settings="{'headers': {'Content-Security-Policy': 'frame-ancestors ${TEXERA_ORIGIN}'}}" \
+  --NotebookApp.base_url="${JUPYTER_BASE_URL}" \
   --NotebookApp.default_url=/tree
diff --git a/notebook-migration-service/src/main/scala/org/apache/texera/service/NotebookMigrationService.scala b/notebook-migration-service/src/main/scala/org/apache/texera/service/NotebookMigrationService.scala
index fe9214b0d15..567cf2b4c69 100644
--- a/notebook-migration-service/src/main/scala/org/apache/texera/service/NotebookMigrationService.scala
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/NotebookMigrationService.scala
@@ -35,6 +35,7 @@ import org.apache.texera.dao.SqlServer
 import org.glassfish.jersey.server.filter.RolesAllowedDynamicFeature
 import java.nio.file.Path
 import org.apache.texera.service.resource.{HealthCheckResource, NotebookMigrationResource}
+import org.apache.texera.service.util.JupyterTokenDeriver
 
 class NotebookMigrationService
     extends Application[NotebookMigrationServiceConfiguration]
@@ -61,6 +62,9 @@ class NotebookMigrationService
       configuration: NotebookMigrationServiceConfiguration,
       environment: Environment
   ): Unit = {
+    // Refuse to boot a misconfigured per-user Jupyter rather than failing per request.
+    JupyterTokenDeriver.validateConfiguration()
+
     // Serve backend at /api
     environment.jersey.setUrlPattern("/api/*")
 
diff --git a/notebook-migration-service/src/main/scala/org/apache/texera/service/resource/NotebookMigrationResource.scala b/notebook-migration-service/src/main/scala/org/apache/texera/service/resource/NotebookMigrationResource.scala
index f048eda3f37..63f6e238886 100644
--- a/notebook-migration-service/src/main/scala/org/apache/texera/service/resource/NotebookMigrationResource.scala
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/resource/NotebookMigrationResource.scala
@@ -35,7 +35,12 @@ import org.apache.texera.dao.jooq.generated.tables.WorkflowVersion
 import java.net.{HttpURLConnection, URL}
 import java.nio.charset.StandardCharsets
 import scala.util.control.NonFatal
-import org.apache.texera.common.config.StorageConfig
+import org.apache.texera.service.util.{
+  JupyterEndpointResolver,
+  JupyterEndpoints,
+  JupyterProbe,
+  JupyterProvisioner
+}
 
 object NotebookMigrationResource extends LazyLogging {
 
@@ -58,6 +63,8 @@ object NotebookMigrationResource extends LazyLogging {
       mapper.createObjectNode().put("success", true).put("deleted", deleted)
     )
 
+  // Also the answer when the user has no Jupyter provisioned: either way there is no
+  // server for them to reach.
   private def jupyterUnavailableResponse: Response =
     Response
       .status(500)
@@ -106,50 +113,14 @@ object NotebookMigrationResource extends LazyLogging {
     }
   }
 
-  // The Jupyter server a request targets. internalUrl is what this service calls, publicUrl
-  // is what the browser loads; they differ once Jupyter is containerized, since the
-  // in-network name does not resolve from the browser. Passed per call so the two can be
-  // made distinct, and so per-user resolution (#7665) can build one of these per uid.
-  final case class JupyterEndpoints(internalUrl: String, publicUrl: String, token: String)
-
-  // Configured default. Process-wide, so this service still targets one Jupyter per process
-  // (the per-user-pod model) and must not be deployed as a shared global instance yet: every
-  // user would get the same Jupyter and token. Per-user resolution is #7665.
-  private val configuredEndpoints = JupyterEndpoints(
-    StorageConfig.jupyterInternalURL,
-    StorageConfig.jupyterPublicURL,
-    StorageConfig.jupyterToken
-  )
-
   // Default notebook name used when a request does not specify one, so a param-less
   // getJupyterIframeURL call reproduces the URL from before this service became stateless.
   private val defaultNotebookName = "notebook.ipynb"
 
-  private def isJupyterAvailable(jupyterUrl: String): Boolean = {
-    var conn: java.net.HttpURLConnection = null
-    try {
-      conn = new java.net.URL(s"$jupyterUrl/api")
-        .openConnection()
-        .asInstanceOf[java.net.HttpURLConnection]
-
-      conn.setRequestMethod("GET")
-      conn.setConnectTimeout(2000)
-      conn.setReadTimeout(2000)
-
-      val status = conn.getResponseCode
-
-      status == 200 || status == 403
-    } catch {
-      case _: Exception => false
-    } finally {
-      if (conn != null) conn.disconnect()
-    }
-  }
-
   // Returns the Jupyter iframe reference URL for the given notebook.
   def getJupyterIframeURL(
       notebookName: String,
-      jupyter: JupyterEndpoints = configuredEndpoints
+      jupyter: JupyterEndpoints = JupyterEndpoints.configured
   ): Response = {
     // notebookName flows into the returned URL, so validate it the same way setNotebook does:
     // block path traversal and keep it to a plain .ipynb filename.
@@ -160,7 +131,7 @@ object NotebookMigrationResource extends LazyLogging {
         .build()
     }
 
-    if (!isJupyterAvailable(jupyter.internalUrl)) {
+    if (!JupyterProbe.isAvailable(jupyter.internalUrl)) {
       return jupyterUnavailableResponse
     }
 
@@ -172,8 +143,8 @@ object NotebookMigrationResource extends LazyLogging {
   }
 
   // Returns the URL of Jupyter
-  def getJupyterURL(jupyter: JupyterEndpoints = configuredEndpoints): Response = {
-    if (!isJupyterAvailable(jupyter.internalUrl)) {
+  def getJupyterURL(jupyter: JupyterEndpoints = JupyterEndpoints.configured): Response = {
+    if (!JupyterProbe.isAvailable(jupyter.internalUrl)) {
       return jupyterUnavailableResponse
     }
 
@@ -181,7 +152,10 @@ object NotebookMigrationResource extends LazyLogging {
   }
 
   // Set the notebook in Jupyter
-  def setNotebook(body: String, jupyter: JupyterEndpoints = configuredEndpoints): Response = {
+  def setNotebook(
+      body: String,
+      jupyter: JupyterEndpoints = JupyterEndpoints.configured
+  ): Response = {
     var conn: HttpURLConnection = null
     try {
       val json = parseBody(body) match {
@@ -202,7 +176,7 @@ object NotebookMigrationResource extends LazyLogging {
           .build()
       }
 
-      if (!isJupyterAvailable(jupyter.internalUrl)) {
+      if (!JupyterProbe.isAvailable(jupyter.internalUrl)) {
         return jupyterUnavailableResponse
       }
 
@@ -271,7 +245,10 @@ object NotebookMigrationResource extends LazyLogging {
   }
 
   // Delete the notebook file from Jupyter's work/ directory:
-  def deleteNotebook(body: String, jupyter: JupyterEndpoints = configuredEndpoints): Response = {
+  def deleteNotebook(
+      body: String,
+      jupyter: JupyterEndpoints = JupyterEndpoints.configured
+  ): Response = {
     var conn: HttpURLConnection = null
     try {
       val json = parseBody(body) match {
@@ -290,7 +267,7 @@ object NotebookMigrationResource extends LazyLogging {
           .build()
       }
 
-      if (!isJupyterAvailable(jupyter.internalUrl)) {
+      if (!JupyterProbe.isAvailable(jupyter.internalUrl)) {
         return jupyterUnavailableResponse
       }
 
@@ -566,6 +543,27 @@ object NotebookMigrationResource extends LazyLogging {
 @Consumes(Array(MediaType.APPLICATION_JSON))
 class NotebookMigrationResource extends LazyLogging {
 
+  // Runs `call` against the caller's own Jupyter, starting one if they have none. The uid
+  // comes from the authenticated session, so a request cannot name another user's server.
+  private def withNewJupyter(user: SessionUser)(call: JupyterEndpoints => Response): Response =
+    respondWith(JupyterProvisioner.ensure(user.getUid), call)
+
+  // As above, but never starts a pod: reading a URL or deleting a file should not bring a
+  // Jupyter into existence for a user who has none.
+  private def withJupyter(user: SessionUser)(call: JupyterEndpoints => Response): Response =
+    respondWith(JupyterEndpointResolver.resolve(user.getUid), call)
+
+  // Visible to the spec so the no-Jupyter branch can be driven directly: the two callers
+  // above resolve against live configuration, which a test cannot flip.
+  private[resource] def respondWith(
+      jupyter: Option[JupyterEndpoints],
+      call: JupyterEndpoints => Response
+  ): Response =
+    jupyter match {
+      case Some(endpoints) => call(endpoints)
+      case None            => NotebookMigrationResource.jupyterUnavailableResponse
+    }
+
   @GET
   @Path("/get-jupyter-iframe-url")
   def getJupyterIframeURL(
@@ -576,28 +574,30 @@ class NotebookMigrationResource extends LazyLogging {
     val name = Option(notebookName)
       .filter(_.nonEmpty)
       .getOrElse(NotebookMigrationResource.defaultNotebookName)
-    NotebookMigrationResource.getJupyterIframeURL(name)
+    withNewJupyter(user) { jupyter =>
+      NotebookMigrationResource.getJupyterIframeURL(name, jupyter)
+    }
   }
 
   @GET
   @Path("/get-jupyter-url")
   def getJupyterURL(@Auth user: SessionUser): Response = {
     logger.info("Getting Jupyter API URL")
-    NotebookMigrationResource.getJupyterURL()
+    withJupyter(user)(NotebookMigrationResource.getJupyterURL)
   }
 
   @POST
   @Path("/set-notebook")
   def setNotebook(body: String, @Auth user: SessionUser): Response = {
     logger.info("Setting notebook")
-    NotebookMigrationResource.setNotebook(body)
+    withNewJupyter(user)(NotebookMigrationResource.setNotebook(body, _))
   }
 
   @POST
   @Path("/delete-notebook")
   def deleteNotebook(body: String, @Auth user: SessionUser): Response = {
     logger.info("Deleting notebook from Jupyter")
-    NotebookMigrationResource.deleteNotebook(body)
+    withJupyter(user)(NotebookMigrationResource.deleteNotebook(body, _))
   }
 
   @POST
diff --git a/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterEndpointResolver.scala b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterEndpointResolver.scala
new file mode 100644
index 00000000000..b4223a8c6ed
--- /dev/null
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterEndpointResolver.scala
@@ -0,0 +1,61 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import org.apache.texera.common.config.{KubernetesConfig, StorageConfig}
+import org.apache.texera.dao.SqlServer
+import org.apache.texera.dao.jooq.generated.tables.daos.UserJupyterDao
+
+/**
+  * Maps a user to the Jupyter their requests should reach.
+  *
+  * The uid always comes from the authenticated session, never from a request body, so one user
+  * can never address another's Jupyter.
+  */
+object JupyterEndpointResolver {
+
+  /**
+    * Endpoints for the user's Jupyter, or None when they have none.
+    *
+    * With per-user Jupyter off, every user resolves to the statically configured server: that
+    * is how the single-node and local-dev deployments run one shared JupyterLab. With it on, a
+    * user with no registry row has nothing provisioned yet, and falling back to the shared
+    * server would hand them somebody else's notebooks.
+    */
+  def resolve(
+      uid: Int,
+      jupyterEnabled: Boolean = KubernetesConfig.jupyterEnabled,
+      fallback: JupyterEndpoints = JupyterEndpoints.configured,
+      tokenSecret: String = StorageConfig.jupyterTokenSecret
+  ): Option[JupyterEndpoints] =
+    if (!jupyterEnabled) Some(fallback)
+    else
+      registrationOf(uid).map(row =>
+        // The token is derived rather than stored, so it is rebuilt here from the uid.
+        JupyterEndpoints(
+          row.getInternalUrl,
+          row.getPublicUrl,
+          JupyterTokenDeriver.derive(uid, tokenSecret)
+        )
+      )
+
+  private def registrationOf(uid: Int) = {
+    val dao = new UserJupyterDao(SqlServer.getInstance().createDSLContext().configuration())
+    Option(dao.fetchOneByUid(uid))
+  }
+}
diff --git a/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterEndpoints.scala b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterEndpoints.scala
new file mode 100644
index 00000000000..5f5ea653b8e
--- /dev/null
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterEndpoints.scala
@@ -0,0 +1,37 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import org.apache.texera.common.config.StorageConfig
+
+/**
+  * The Jupyter server a request targets. internalUrl is what the service calls, publicUrl is
+  * what the browser loads; they differ once Jupyter is containerized, since the in-network
+  * name does not resolve from the browser.
+  */
+final case class JupyterEndpoints(internalUrl: String, publicUrl: String, token: String)
+
+object JupyterEndpoints {
+
+  // The single Jupyter from static config, used while per-user provisioning is off.
+  val configured: JupyterEndpoints = JupyterEndpoints(
+    StorageConfig.jupyterInternalURL,
+    StorageConfig.jupyterPublicURL,
+    StorageConfig.jupyterToken
+  )
+}
diff --git a/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterKubernetesClient.scala b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterKubernetesClient.scala
new file mode 100644
index 00000000000..8afad3f7617
--- /dev/null
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterKubernetesClient.scala
@@ -0,0 +1,122 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import io.fabric8.kubernetes.api.model.{
+  EnvVarBuilder,
+  Pod,
+  PodBuilder,
+  Quantity,
+  ResourceRequirementsBuilder
+}
+import io.fabric8.kubernetes.client.KubernetesClientBuilder
+import org.apache.texera.common.config.KubernetesConfig
+
+/**
+  * Thin wrapper over the fabric8 client for per-user JupyterLab pods, mirroring the computing
+  * unit's KubernetesClient. The fabric8 client is a constructor parameter rather than a global
+  * so tests can exercise the naming and addressing without a live cluster.
+  */
+class JupyterKubernetesClient(client: io.fabric8.kubernetes.client.KubernetesClient) {
+
+  private val namespace: String = KubernetesConfig.jupyterNamespace
+  private val podNamePrefix = "jupyter"
+
+  def generatePodName(uid: Int): String = s"$podNamePrefix-$uid"
+
+  /** The in-cluster address of a user's pod, resolvable via the headless service. */
+  def generatePodURI(uid: Int): String =
+    s"${generatePodName(uid)}.${KubernetesConfig.jupyterServiceName}.$namespace.svc.cluster.local:${KubernetesConfig.jupyterPortNumber}"
+
+  /**
+    * Path a user's Jupyter serves under. The uid is in the path because the browser cannot
+    * present Texera credentials on the requests Jupyter's own scripts make, so the gateway
+    * has to read the owner out of the URL instead.
+    */
+  def basePathFor(uid: Int): String = s"${KubernetesConfig.jupyterBaseUrl.stripSuffix("/")}/$uid"
+
+  def podExists(uid: Int): Boolean = getPodByName(generatePodName(uid)).isDefined
+
+  def getPodByName(podName: String): Option[Pod] =
+    Option(client.pods().inNamespace(namespace).withName(podName).get())
+
+  /**
+    * Starts a user's JupyterLab. The token is passed as JUPYTER_TOKEN, which is what the image's
+    * start-texera-jupyter.sh reads, so each pod ends up with its owner's token and no other.
+    * Hostname and subdomain are what make generatePodURI resolve.
+    */
+  def createPod(uid: Int, token: String): Pod = {
+    val podName = generatePodName(uid)
+
+    val resources = new ResourceRequirementsBuilder()
+      .addToLimits("cpu", new Quantity(KubernetesConfig.jupyterCpuLimit))
+      .addToLimits("memory", new Quantity(KubernetesConfig.jupyterMemoryLimit))
+      .build()
+
+    val pod = new PodBuilder()
+      .withNewMetadata()
+      .withName(podName)
+      .withNamespace(namespace)
+      .addToLabels("type", "jupyter")
+      .addToLabels("uid", uid.toString)
+      .addToLabels("name", podName)
+      .endMetadata()
+      .withNewSpec()
+      .addNewContainer()
+      .withName("jupyter")
+      .withImage(KubernetesConfig.jupyterImageName)
+      .withImagePullPolicy(KubernetesConfig.computingUnitImagePullPolicy)
+      .addNewPort()
+      .withContainerPort(KubernetesConfig.jupyterPortNumber)
+      .endPort()
+      .withEnv(
+        new EnvVarBuilder().withName("JUPYTER_TOKEN").withValue(token).build(),
+        new EnvVarBuilder()
+          .withName("JUPYTER_BASE_URL")
+          .withValue(s"${basePathFor(uid)}/")
+          .build(),
+        // Drives the pod's iframe CSP and its postMessage origin check. Empty is ignored by
+        // the image, which then keeps its local-development default.
+        new EnvVarBuilder()
+          .withName("TEXERA_ORIGIN")
+          .withValue(KubernetesConfig.jupyterTexeraOrigin)
+          .build()
+      )
+      .withResources(resources)
+      .endContainer()
+      .withHostname(podName)
+      .withSubdomain(KubernetesConfig.jupyterServiceName)
+      .endSpec()
+      .build()
+
+    client.resource(pod).inNamespace(namespace).create()
+  }
+
+  def deletePod(uid: Int): Unit =
+    client.pods().inNamespace(namespace).withName(generatePodName(uid)).delete()
+}
+
+object JupyterKubernetesClient {
+
+  /**
+    * Built on demand rather than at object initialisation: the single-node and local-dev
+    * deployments have no cluster to build a client against, and never provision.
+    */
+  def inCluster: JupyterKubernetesClient =
+    new JupyterKubernetesClient(new KubernetesClientBuilder().build())
+}
diff --git a/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterProbe.scala b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterProbe.scala
new file mode 100644
index 00000000000..c60e475af17
--- /dev/null
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterProbe.scala
@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import java.net.{HttpURLConnection, URL}
+
+/** Liveness check for a Jupyter server. */
+object JupyterProbe {
+
+  private val timeoutMillis = 2000
+
+  /**
+    * Whether Jupyter answers on `internalUrl`. /api returns the server version without a
+    * token, so 403 counts as reachable: the server is up and merely refusing the request.
+    */
+  def isAvailable(internalUrl: String): Boolean = {
+    var conn: HttpURLConnection = null
+    try {
+      conn = new URL(s"$internalUrl/api").openConnection().asInstanceOf[HttpURLConnection]
+      conn.setRequestMethod("GET")
+      conn.setConnectTimeout(timeoutMillis)
+      conn.setReadTimeout(timeoutMillis)
+      val status = conn.getResponseCode
+      status == 200 || status == 403
+    } catch {
+      case _: Exception => false
+    } finally {
+      if (conn != null) conn.disconnect()
+    }
+  }
+}
diff --git a/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterProvisioner.scala b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterProvisioner.scala
new file mode 100644
index 00000000000..0431dc9eea2
--- /dev/null
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterProvisioner.scala
@@ -0,0 +1,145 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import com.typesafe.scalalogging.LazyLogging
+import org.apache.texera.common.config.{KubernetesConfig, StorageConfig}
+import org.apache.texera.dao.SqlServer
+import org.apache.texera.dao.jooq.generated.tables.daos.UserJupyterDao
+import org.apache.texera.dao.jooq.generated.tables.pojos.UserJupyter
+import org.jooq.exception.DataAccessException
+
+import scala.util.control.NonFatal
+
+/**
+  * Brings a user's JupyterLab into existence and registers where it lives.
+  *
+  * Dependencies are constructor parameters so the provisioning logic can be tested without a
+  * cluster; the companion object binds the production ones.
+  */
+class JupyterProvisioner(
+    kubernetesClient: => JupyterKubernetesClient,
+    isReachable: String => Boolean,
+    publicUrlTemplate: String,
+    readinessTimeoutMillis: Long,
+    readinessPollMillis: Long
+) extends LazyLogging {
+
+  // By-name above, forced once here, so no client is built unless a provision happens.
+  private lazy val kubernetes = kubernetesClient
+
+  /**
+    * The user's Jupyter, starting one if they have none. None means it could not be made
+    * ready, which callers report the same as an unreachable server.
+    *
+    * A registered pod that no longer answers is discarded and rebuilt: the row would otherwise
+    * outlive the pod and point every later request at nothing.
+    */
+  def ensure(
+      uid: Int,
+      jupyterEnabled: Boolean = KubernetesConfig.jupyterEnabled,
+      fallback: JupyterEndpoints = JupyterEndpoints.configured,
+      tokenSecret: String = StorageConfig.jupyterTokenSecret
+  ): Option[JupyterEndpoints] = {
+    if (!jupyterEnabled) return Some(fallback)
+
+    val token = JupyterTokenDeriver.derive(uid, tokenSecret)
+    JupyterEndpointResolver.resolve(uid, jupyterEnabled = true, tokenSecret = tokenSecret) match {
+      case Some(endpoints) if isReachable(endpoints.internalUrl) => Some(endpoints)
+      case Some(endpoints) =>
+        logger.warn(
+          s"Jupyter for user $uid is registered at ${endpoints.internalUrl} but "
+            + "unreachable; rebuilding it"
+        )
+        discard(uid)
+        provision(uid, token)
+      case None => provision(uid, token)
+    }
+  }
+
+  private def provision(uid: Int, token: String): Option[JupyterEndpoints] = {
+    // Jupyter serves every endpoint under its base path, /api included, so the recorded
+    // address has to carry it or each later call lands on a 404.
+    val internalUrl = s"http://${kubernetes.generatePodURI(uid)}${kubernetes.basePathFor(uid)}"
+    val endpoints = JupyterEndpoints(internalUrl, publicUrlFor(uid, internalUrl), token)
+    try {
+      if (!kubernetes.podExists(uid)) kubernetes.createPod(uid, token)
+      if (!waitUntilReachable(internalUrl)) {
+        logger.error(s"Jupyter for user $uid did not become ready; removing the pod")
+        kubernetes.deletePod(uid)
+        None
+      } else {
+        register(endpoints, uid)
+        Some(endpoints)
+      }
+    } catch {
+      case NonFatal(e) =>
+        logger.error(s"Failed to provision Jupyter for user $uid", e)
+        None
+    }
+  }
+
+  /** Browser-facing address; the in-cluster name does not resolve from the browser. */
+  private def publicUrlFor(uid: Int, internalUrl: String): String =
+    if (publicUrlTemplate.isEmpty) internalUrl
+    else publicUrlTemplate.replace("{uid}", uid.toString)
+
+  private def waitUntilReachable(internalUrl: String): Boolean = {
+    val deadline = System.currentTimeMillis() + readinessTimeoutMillis
+    var ready = isReachable(internalUrl)
+    while (!ready && System.currentTimeMillis() < deadline) {
+      Thread.sleep(readinessPollMillis)
+      ready = isReachable(internalUrl)
+    }
+    ready
+  }
+
+  private def register(endpoints: JupyterEndpoints, uid: Int): Unit = {
+    val row = new UserJupyter
+    row.setUid(uid)
+    row.setInternalUrl(endpoints.internalUrl)
+    row.setPublicUrl(endpoints.publicUrl)
+    try dao().insert(row)
+    catch {
+      // Two concurrent first requests can both provision. uid is the primary key, so the
+      // loser trips 23505; the winner's row holds the same uid-derived addresses, so leaving
+      // it in place is correct.
+      case e: DataAccessException if e.sqlState == "23505" =>
+        logger.info(s"Jupyter for user $uid was registered concurrently; keeping that row")
+    }
+  }
+
+  private def discard(uid: Int): Unit = {
+    try kubernetes.deletePod(uid)
+    catch { case NonFatal(e) => logger.warn(s"Could not delete stale Jupyter pod for $uid", e) }
+    dao().deleteById(uid)
+  }
+
+  private def dao() = new UserJupyterDao(SqlServer.getInstance().createDSLContext().configuration())
+}
+
+// A pod is scheduled, pulled and started before Jupyter answers, so the first request after
+// provisioning waits rather than failing.
+object JupyterProvisioner
+    extends JupyterProvisioner(
+      JupyterKubernetesClient.inCluster,
+      JupyterProbe.isAvailable,
+      KubernetesConfig.jupyterPublicUrlTemplate,
+      readinessTimeoutMillis = 60000,
+      readinessPollMillis = 1000
+    )
diff --git a/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterTokenDeriver.scala b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterTokenDeriver.scala
new file mode 100644
index 00000000000..ea6b5135e72
--- /dev/null
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterTokenDeriver.scala
@@ -0,0 +1,59 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import org.apache.texera.common.config.{KubernetesConfig, StorageConfig}
+
+import javax.crypto.Mac
+import javax.crypto.spec.SecretKeySpec
+import java.nio.charset.StandardCharsets.UTF_8
+
+/**
+  * Derives each user's JupyterLab token from a server-held secret instead of storing one.
+  * The value is stable for a uid until the secret changes, so any replica of this service
+  * derives the same token and no credential is kept at rest.
+  */
+object JupyterTokenDeriver {
+
+  private val algorithm = "HmacSHA256"
+
+  // 128 bits of the digest, which is ample for a token and keeps the URL short.
+  private val tokenLength = 32
+
+  def derive(uid: Int, secret: String = StorageConfig.jupyterTokenSecret): String = {
+    require(secret.nonEmpty, "cannot derive a Jupyter token from an empty secret")
+    val mac = Mac.getInstance(algorithm)
+    mac.init(new SecretKeySpec(secret.getBytes(UTF_8), algorithm))
+    mac.doFinal(uid.toString.getBytes(UTF_8)).map("%02x".format(_)).mkString.take(tokenLength)
+  }
+
+  /**
+    * Refuses to start per-user Jupyter without a secret: an empty key is public, so anyone
+    * could derive another user's token. Only enforced when the feature is on, so the
+    * single-node and local-dev deployments are unaffected.
+    */
+  def validateConfiguration(
+      jupyterEnabled: Boolean = KubernetesConfig.jupyterEnabled,
+      secret: String = StorageConfig.jupyterTokenSecret
+  ): Unit =
+    if (jupyterEnabled && secret.isEmpty) {
+      throw new IllegalStateException(
+        "kubernetes.jupyter-enabled requires a non-empty storage.jupyter.token-secret"
+      )
+    }
+}
diff --git a/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala b/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala
index 10d5829b874..d185fa44361 100644
--- a/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala
+++ b/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala
@@ -22,21 +22,31 @@ package org.apache.texera.service.resource
 import jakarta.ws.rs.core.Response
 import org.apache.texera.auth.SessionUser
 import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.service.util.{
+  JupyterEndpointResolver,
+  JupyterEndpoints,
+  JupyterKubernetesClient,
+  JupyterProvisioner,
+  JupyterTokenDeriver
+}
 import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, UserRoleEnum}
 import org.apache.texera.dao.jooq.generated.tables.Notebook.NOTEBOOK
 import org.apache.texera.dao.jooq.generated.tables.User.USER
+import org.apache.texera.dao.jooq.generated.tables.UserJupyter.USER_JUPYTER
 import org.apache.texera.dao.jooq.generated.tables.Workflow.WORKFLOW
 import org.apache.texera.dao.jooq.generated.tables.WorkflowNotebookMapping.WORKFLOW_NOTEBOOK_MAPPING
 import org.apache.texera.dao.jooq.generated.tables.WorkflowUserAccess.WORKFLOW_USER_ACCESS
 import org.apache.texera.dao.jooq.generated.tables.WorkflowVersion.WORKFLOW_VERSION
 import org.apache.texera.dao.jooq.generated.tables.daos.{
   UserDao,
+  UserJupyterDao,
   WorkflowDao,
   WorkflowUserAccessDao,
   WorkflowVersionDao
 }
 import org.apache.texera.dao.jooq.generated.tables.pojos.{
   User,
+  UserJupyter,
   Workflow,
   WorkflowUserAccess,
   WorkflowVersion
@@ -49,6 +59,8 @@ import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
 import com.sun.net.httpserver.HttpServer
 
 import java.net.InetSocketAddress
+
+import scala.jdk.CollectionConverters._
 import java.sql.Timestamp
 import java.util.UUID
 
@@ -152,6 +164,7 @@ class NotebookMigrationResourceSpec
       .where(WORKFLOW_VERSION.WID.eq(testWid))
       .execute()
     getDSLContext.deleteFrom(WORKFLOW).where(WORKFLOW.WID.eq(testWid)).execute()
+    getDSLContext.deleteFrom(USER_JUPYTER).execute()
     getDSLContext.deleteFrom(USER).where(USER.EMAIL.in(writerEmail, readerEmail)).execute()
   }
 
@@ -543,12 +556,309 @@ class NotebookMigrationResourceSpec
   // gets. 192.0.2.0/24 is TEST-NET-1 (RFC 5737) and routes nowhere, so a call that wrongly
   // dials the public URL fails rather than silently passing. Numeric on purpose: a hostname
   // would go through the resolver, which setConnectTimeout does not bound.
-  private val splitEndpoints = NotebookMigrationResource.JupyterEndpoints(
+  private val splitEndpoints = JupyterEndpoints(
     internalUrl = "http://localhost:9100",
     publicUrl = "http://192.0.2.1:1234",
     token = "texera"
   )
 
+  // -- per-user resolution ----------------------------------------------------
+
+  // Registers a Jupyter for `uid`, standing in for a provisioned pod.
+  private def registerJupyter(
+      uid: Integer,
+      internalUrl: String = "http://localhost:9100",
+      publicUrl: String = "http://192.0.2.1:1234"
+  ): Unit = {
+    val row = new UserJupyter
+    row.setUid(uid)
+    row.setInternalUrl(internalUrl)
+    row.setPublicUrl(publicUrl)
+    new UserJupyterDao(getDSLContext.configuration()).insert(row)
+  }
+
+  private val specSecret = "resolver-spec-secret"
+
+  "JupyterEndpointResolver" should "resolve every user to the configured Jupyter while the feature is off" in {
+    // How single-node and local dev run: one shared JupyterLab, no registry rows.
+    JupyterEndpointResolver.resolve(writerUid, jupyterEnabled = false) shouldBe Some(
+      JupyterEndpoints.configured
+    )
+  }
+
+  it should "resolve a registered user to their own Jupyter" in {
+    registerJupyter(writerUid, internalUrl = "http://jupyter-1:8888")
+    val resolved =
+      JupyterEndpointResolver.resolve(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+    resolved.map(_.internalUrl) shouldBe Some("http://jupyter-1:8888")
+    resolved.map(_.publicUrl) shouldBe Some("http://192.0.2.1:1234")
+  }
+
+  it should "return None for an unregistered user rather than falling back to the shared Jupyter" in {
+    // The isolation property: falling back here would hand an unprovisioned user somebody
+    // else's notebooks, which is the whole point of resolving per user.
+    JupyterEndpointResolver.resolve(
+      writerUid,
+      jupyterEnabled = true,
+      tokenSecret = specSecret
+    ) shouldBe None
+  }
+
+  it should "never return one user's Jupyter to another" in {
+    registerJupyter(writerUid, internalUrl = "http://jupyter-writer:8888")
+    JupyterEndpointResolver
+      .resolve(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+      .map(_.internalUrl) shouldBe Some("http://jupyter-writer:8888")
+    JupyterEndpointResolver.resolve(
+      readerUid,
+      jupyterEnabled = true,
+      tokenSecret = specSecret
+    ) shouldBe None
+  }
+
+  it should "derive the registered user's token rather than reading one from the row" in {
+    // No token column exists, so the resolver has to rebuild it from the uid.
+    registerJupyter(writerUid)
+    JupyterEndpointResolver
+      .resolve(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+      .map(_.token) shouldBe Some(JupyterTokenDeriver.derive(writerUid, specSecret))
+  }
+
+  it should "give two registered users different tokens" in {
+    registerJupyter(writerUid)
+    registerJupyter(readerUid)
+    val writerToken = JupyterEndpointResolver
+      .resolve(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+      .map(_.token)
+    val readerToken = JupyterEndpointResolver
+      .resolve(readerUid, jupyterEnabled = true, tokenSecret = specSecret)
+      .map(_.token)
+    writerToken should not be readerToken
+  }
+
+  "the resource class" should "report Jupyter unavailable when the caller has none" in {
+    // What an unprovisioned user gets: never a fall back to somebody else's Jupyter.
+    val response = new NotebookMigrationResource()
+      .respondWith(None, _ => fail("must not call through without a Jupyter"))
+    response.getStatus shouldBe 500
+    response.getEntity.toString should include("Cannot connect to Jupyter server")
+  }
+
+  it should "call through to the endpoint when the caller has a Jupyter" in {
+    val response = new NotebookMigrationResource()
+      .respondWith(Some(splitEndpoints), jupyter => Response.ok(jupyter.internalUrl).build())
+    response.getEntity.toString shouldBe "http://localhost:9100"
+  }
+
+  // -- provisioning -----------------------------------------------------------
+
+  // Records what would have been asked of Kubernetes, so the provisioning logic runs without
+  // a cluster. Subclassing rather than mocking keeps the real naming and addressing.
+  private class StubKubernetes extends JupyterKubernetesClient(null) {
+    var created: List[(Int, String)] = Nil
+    var deleted: List[Int] = Nil
+    var alreadyExists = false
+    var failCreate = false
+    var failDelete = false
+    override def podExists(uid: Int): Boolean = alreadyExists
+    override def createPod(uid: Int, token: String) = {
+      if (failCreate) throw new RuntimeException("cluster refused the pod")
+      created ::= ((uid, token))
+      null
+    }
+    override def deletePod(uid: Int): Unit = {
+      deleted ::= uid
+      if (failDelete) throw new RuntimeException("pod already gone")
+    }
+  }
+
+  // Short windows so the "never ready" path does not sit in a real timeout.
+  private def provisionerFor(
+      kubernetes: JupyterKubernetesClient,
+      reachable: String => Boolean,
+      publicUrlTemplate: String = ""
+  ) =
+    new JupyterProvisioner(
+      kubernetes,
+      reachable,
+      publicUrlTemplate,
+      readinessTimeoutMillis = 50,
+      readinessPollMillis = 10
+    )
+
+  private def registeredUids(): List[Integer] =
+    getDSLContext
+      .select(USER_JUPYTER.UID)
+      .from(USER_JUPYTER)
+      .fetchInto(classOf[Integer])
+      .asScala
+      .toList
+
+  "JupyterProvisioner.ensure" should "return the configured Jupyter and start nothing while the feature is off" in {
+    val kubernetes = new StubKubernetes
+    val result = provisionerFor(kubernetes, _ => true).ensure(writerUid, jupyterEnabled = false)
+    result shouldBe Some(JupyterEndpoints.configured)
+    kubernetes.created shouldBe empty
+    registeredUids() shouldBe empty
+  }
+
+  it should "start and register a Jupyter for a user who has none" in {
+    val kubernetes = new StubKubernetes
+    val result = provisionerFor(kubernetes, _ => true)
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    kubernetes.created.map(_._1) shouldBe List(writerUid.intValue())
+    registeredUids() shouldBe List(writerUid)
+    result.map(_.internalUrl) shouldBe Some(
+      s"http://${kubernetes.generatePodURI(writerUid)}${kubernetes.basePathFor(writerUid)}"
+    )
+  }
+
+  it should "give the pod the user's own derived token" in {
+    // What makes one user's token useless against another's Jupyter.
+    val kubernetes = new StubKubernetes
+    provisionerFor(kubernetes, _ => true)
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+    kubernetes.created.map(_._2) shouldBe List(JupyterTokenDeriver.derive(writerUid, specSecret))
+  }
+
+  it should "reuse a registered Jupyter that still answers" in {
+    registerJupyter(writerUid)
+    val kubernetes = new StubKubernetes
+    val result = provisionerFor(kubernetes, _ => true)
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    kubernetes.created shouldBe empty
+    result.map(_.internalUrl) shouldBe Some("http://localhost:9100")
+  }
+
+  it should "rebuild a registered Jupyter whose pod is gone" in {
+    // The row would otherwise outlive the pod and point every later request at nothing.
+    registerJupyter(writerUid, internalUrl = "http://stale:8888")
+    val kubernetes = new StubKubernetes
+    val result = provisionerFor(kubernetes, url => url != "http://stale:8888")
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    kubernetes.deleted shouldBe List(writerUid.intValue())
+    kubernetes.created.map(_._1) shouldBe List(writerUid.intValue())
+    result.map(_.internalUrl) shouldBe Some(
+      s"http://${kubernetes.generatePodURI(writerUid)}${kubernetes.basePathFor(writerUid)}"
+    )
+  }
+
+  it should "register nothing and clean up when the pod never becomes ready" in {
+    val kubernetes = new StubKubernetes
+    val result = provisionerFor(kubernetes, _ => false)
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    result shouldBe None
+    kubernetes.deleted shouldBe List(writerUid.intValue())
+    registeredUids() shouldBe empty
+  }
+
+  it should "record the user's own base path in the internal address" in {
+    // Jupyter serves /api under its base path too, so an address without the prefix would
+    // make every later probe and contents call 404.
+    val kubernetes = new StubKubernetes
+    val result = provisionerFor(kubernetes, _ => true)
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    result.map(_.internalUrl) shouldBe Some(
+      s"http://${kubernetes.generatePodURI(writerUid)}${kubernetes.basePathFor(writerUid)}"
+    )
+  }
+
+  it should "build the public URL from the configured template" in {
+    val kubernetes = new StubKubernetes
+    val result = provisionerFor(kubernetes, _ => true, "https://texera.example.com/jupyter/{uid}")
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+    result.map(_.publicUrl) shouldBe Some(s"https://texera.example.com/jupyter/$writerUid")
+  }
+
+  it should "adopt an existing pod instead of creating a second one" in {
+    // A pod can outlive its row, so provisioning must be idempotent on the Kubernetes side.
+    val kubernetes = new StubKubernetes
+    kubernetes.alreadyExists = true
+    val result = provisionerFor(kubernetes, _ => true)
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    kubernetes.created shouldBe empty
+    result should not be empty
+    registeredUids() shouldBe List(writerUid)
+  }
+
+  it should "reuse one Kubernetes client across calls" in {
+    // The client is built lazily and held, so a second request must not construct another.
+    val kubernetes = new StubKubernetes
+    val provisioner = provisionerFor(kubernetes, _ => true)
+    provisioner.ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+    provisioner.ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    // Second call finds the row it just wrote, so it provisions once in total.
+    kubernetes.created.map(_._1) shouldBe List(writerUid.intValue())
+    registeredUids() shouldBe List(writerUid)
+  }
+
+  it should "report unavailable when the cluster refuses to create the pod" in {
+    val kubernetes = new StubKubernetes
+    kubernetes.failCreate = true
+    val result = provisionerFor(kubernetes, _ => true)
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    result shouldBe None
+    registeredUids() shouldBe empty
+  }
+
+  it should "still rebuild when the stale pod cannot be deleted" in {
+    // The pod may already be gone, which is the state the delete was trying to reach.
+    registerJupyter(writerUid, internalUrl = "http://stale:8888")
+    val kubernetes = new StubKubernetes
+    kubernetes.failDelete = true
+    val result = provisionerFor(kubernetes, url => url != "http://stale:8888")
+      .ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    kubernetes.created.map(_._1) shouldBe List(writerUid.intValue())
+    result.map(_.internalUrl) shouldBe Some(
+      s"http://${kubernetes.generatePodURI(writerUid)}${kubernetes.basePathFor(writerUid)}"
+    )
+  }
+
+  it should "report unavailable when registration fails for a reason other than a race" in {
+    // A uid with no user row violates the foreign key. Only a duplicate primary key means
+    // "another request won"; anything else has to surface rather than be swallowed.
+    val orphanUid = 999999
+    val kubernetes = new StubKubernetes
+    val result = provisionerFor(kubernetes, _ => true)
+      .ensure(orphanUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    result shouldBe None
+    registeredUids() shouldBe empty
+  }
+
+  it should "keep the winning row when two requests provision at once" in {
+    // The readiness probe runs just before the insert, so registering there stands in for a
+    // concurrent request winning the race.
+    val kubernetes = new StubKubernetes
+    val racing = provisionerFor(
+      kubernetes,
+      _ => { if (registeredUids().isEmpty) registerJupyter(writerUid); true }
+    )
+    val result = racing.ensure(writerUid, jupyterEnabled = true, tokenSecret = specSecret)
+
+    result should not be empty
+    registeredUids() shouldBe List(writerUid)
+  }
+
+  it should "fall back to the configured endpoints when none are passed" in {
+    // The defaulted parameter is what keeps direct object calls working for callers that
+    // have no per-user endpoints to hand in.
+    withFakeJupyter(contentsStatus = 201) {
+      val response = NotebookMigrationResource.getJupyterURL()
+      response.getStatus shouldBe Response.Status.OK.getStatusCode
+      response.getEntity.toString should include(JupyterEndpoints.configured.publicUrl)
+    }
+  }
+
   "the internal/public URL split" should "dial the internal URL and return only the public one" in {
     withFakeJupyter(contentsStatus = 201) {
       val urlResp = NotebookMigrationResource.getJupyterURL(splitEndpoints)
@@ -593,7 +903,7 @@ class NotebookMigrationResourceSpec
     // rescue an unreachable internal one.
     withFakeJupyter(contentsStatus = 201) {
       // Port 9 on loopback: refused immediately, so this fails fast and without DNS.
-      val swapped = NotebookMigrationResource.JupyterEndpoints(
+      val swapped = JupyterEndpoints(
         internalUrl = "http://127.0.0.1:9",
         publicUrl = "http://localhost:9100",
         token = "texera"
diff --git a/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterEndpointsSpec.scala b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterEndpointsSpec.scala
new file mode 100644
index 00000000000..537c94dad3a
--- /dev/null
+++ b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterEndpointsSpec.scala
@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import org.apache.texera.common.config.StorageConfig
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class JupyterEndpointsSpec extends AnyFlatSpec with Matchers {
+
+  private val endpoints = JupyterEndpoints("http://internal:8888", "http://public", "tok")
+
+  "JupyterEndpoints.configured" should "mirror the static Jupyter configuration" in {
+    JupyterEndpoints.configured shouldBe JupyterEndpoints(
+      StorageConfig.jupyterInternalURL,
+      StorageConfig.jupyterPublicURL,
+      StorageConfig.jupyterToken
+    )
+  }
+
+  "JupyterEndpoints" should "compare by value" in {
+    endpoints shouldBe JupyterEndpoints("http://internal:8888", "http://public", "tok")
+    endpoints.hashCode shouldBe
+      JupyterEndpoints("http://internal:8888", "http://public", "tok").hashCode
+  }
+
+  it should "differ when any field differs" in {
+    // The token is part of identity: two users share a URL in the fallback case but never a token.
+    endpoints should not be endpoints.copy(token = "other")
+    endpoints should not be endpoints.copy(internalUrl = "http://other:8888")
+    endpoints should not be endpoints.copy(publicUrl = "http://other")
+  }
+
+  it should "not equal a value of another type" in {
+    endpoints should not be "http://internal:8888"
+    endpoints.toString should include("http://internal:8888")
+  }
+
+  it should "destructure into its three parts" in {
+    val JupyterEndpoints(internal, public, token) = endpoints
+    internal shouldBe "http://internal:8888"
+    public shouldBe "http://public"
+    token shouldBe "tok"
+  }
+}
diff --git a/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterKubernetesClientSpec.scala b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterKubernetesClientSpec.scala
new file mode 100644
index 00000000000..c6d1a6c276f
--- /dev/null
+++ b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterKubernetesClientSpec.scala
@@ -0,0 +1,213 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, PodList}
+import io.fabric8.kubernetes.client.dsl.{
+  MixedOperation,
+  NamespaceableResource,
+  NonNamespaceOperation,
+  PodResource,
+  Resource
+}
+import io.fabric8.kubernetes.client.{KubernetesClient => Fabric8Client}
+import org.apache.texera.common.config.KubernetesConfig
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify, when}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.jdk.CollectionConverters._
+
+/**
+  * Two layers: the pure naming and addressing, which is what the registry stores and the
+  * service dials, and the thin fabric8 wrappers driven through a Mockito-stubbed client so
+  * the pod spec can be asserted without a live cluster.
+  */
+class JupyterKubernetesClientSpec extends AnyFlatSpec with Matchers {
+
+  private val namespace = KubernetesConfig.jupyterNamespace
+
+  private val bare = new JupyterKubernetesClient(null)
+
+  // fabric8's fluent API returns type variables, so RETURNS_DEEP_STUBS cannot be used and
+  // each step of the chain is mocked explicitly. Mirrors the computing unit's spec.
+  private def stubbedPods(existing: Pod): (Fabric8Client, PodResource) = {
+    val client = mock(classOf[Fabric8Client])
+    val mixed = mock(classOf[MixedOperation[_, _, _]])
+      .asInstanceOf[MixedOperation[Pod, PodList, PodResource]]
+    val inNamespace = mock(classOf[NonNamespaceOperation[_, _, _]])
+      .asInstanceOf[NonNamespaceOperation[Pod, PodList, PodResource]]
+    val podResource = mock(classOf[PodResource])
+    when(client.pods()).thenReturn(mixed)
+    when(mixed.inNamespace(namespace)).thenReturn(inNamespace)
+    when(inNamespace.withName(org.mockito.ArgumentMatchers.anyString())).thenReturn(podResource)
+    when(podResource.get()).thenReturn(existing)
+    (client, podResource)
+  }
+
+  // -- naming and addressing --------------------------------------------------
+
+  "generatePodName" should "namespace the pod by uid" in {
+    bare.generatePodName(7) shouldBe "jupyter-7"
+  }
+
+  it should "give every user a distinct pod name" in {
+    (1 to 50).map(bare.generatePodName).distinct.size shouldBe 50
+  }
+
+  "generatePodURI" should "address the pod through the headless service" in {
+    // Must match the pod's hostname.subdomain, or the name does not resolve in-cluster.
+    bare.generatePodURI(7) shouldBe
+      s"jupyter-7.${KubernetesConfig.jupyterServiceName}.$namespace" +
+        s".svc.cluster.local:${KubernetesConfig.jupyterPortNumber}"
+  }
+
+  it should "carry the configured port" in {
+    bare.generatePodURI(7) should endWith(s":${KubernetesConfig.jupyterPortNumber}")
+  }
+
+  // -- lookups ---------------------------------------------------------------
+
+  "getPodByName" should "return the pod when one exists" in {
+    val pod = new PodBuilder().withNewMetadata().withName("jupyter-7").endMetadata().build()
+    val (client, _) = stubbedPods(pod)
+    new JupyterKubernetesClient(client).getPodByName("jupyter-7") shouldBe Some(pod)
+  }
+
+  it should "return None when the pod is absent" in {
+    val (client, _) = stubbedPods(null)
+    new JupyterKubernetesClient(client).getPodByName("jupyter-7") shouldBe None
+  }
+
+  "podExists" should "report true for a live pod and false for a missing one" in {
+    val pod = new PodBuilder().withNewMetadata().withName("jupyter-7").endMetadata().build()
+    new JupyterKubernetesClient(stubbedPods(pod)._1).podExists(7) shouldBe true
+    new JupyterKubernetesClient(stubbedPods(null)._1).podExists(7) shouldBe false
+  }
+
+  "deletePod" should "delete the user's own pod by name" in {
+    val (client, podResource) = stubbedPods(null)
+    new JupyterKubernetesClient(client).deletePod(7)
+    verify(client.pods().inNamespace(namespace)).withName("jupyter-7")
+    verify(podResource).delete()
+  }
+
+  // -- pod spec --------------------------------------------------------------
+
+  // Captures the pod handed to fabric8, so every field the deployment depends on is asserted.
+  private def createdPod(uid: Int, token: String): Pod = {
+    val client = mock(classOf[Fabric8Client])
+    val namespaceable = mock(classOf[NamespaceableResource[_]])
+      .asInstanceOf[NamespaceableResource[Pod]]
+    val resource = mock(classOf[Resource[_]]).asInstanceOf[Resource[Pod]]
+    val captor = ArgumentCaptor.forClass(classOf[Pod])
+    when(client.resource(captor.capture())).thenReturn(namespaceable)
+    when(namespaceable.inNamespace(namespace)).thenReturn(resource)
+    when(resource.create()).thenReturn(null)
+    new JupyterKubernetesClient(client).createPod(uid, token)
+    captor.getValue
+  }
+
+  "createPod" should "name and namespace the pod for its owner" in {
+    val pod = createdPod(7, "tok")
+    pod.getMetadata.getName shouldBe "jupyter-7"
+    pod.getMetadata.getNamespace shouldBe namespace
+  }
+
+  it should "label the pod so the headless service and the owner are identifiable" in {
+    val labels = createdPod(7, "tok").getMetadata.getLabels.asScala
+    labels("type") shouldBe "jupyter"
+    labels("uid") shouldBe "7"
+    labels("name") shouldBe "jupyter-7"
+  }
+
+  it should "pass the owner's token as JUPYTER_TOKEN" in {
+    // The image's start-texera-jupyter.sh reads this, so it is what isolates one user's
+    // Jupyter from another's.
+    val env = createdPod(7, "derived-token").getSpec.getContainers.asScala.head.getEnv.asScala
+    env.map(_.getName) should contain("JUPYTER_TOKEN")
+    env.find(_.getName == "JUPYTER_TOKEN").map(_.getValue) shouldBe Some("derived-token")
+  }
+
+  it should "tell the pod which base path it serves under" in {
+    // The image passes this to --NotebookApp.base_url. Jupyter wants a trailing slash, and
+    // without the prefix a path-routed deployment serves every endpoint from the wrong place.
+    val env = createdPod(7, "tok").getSpec.getContainers.asScala.head.getEnv.asScala
+    env.find(_.getName == "JUPYTER_BASE_URL").map(_.getValue) shouldBe Some("/jupyter/7/")
+  }
+
+  it should "tell the pod which Texera origin may frame it" in {
+    // Without this the pod keeps the image's local-development default and its CSP
+    // frame-ancestors blocks the real deployment from embedding it.
+    val env = createdPod(7, "tok").getSpec.getContainers.asScala.head.getEnv.asScala
+    env.find(_.getName == "TEXERA_ORIGIN").map(_.getValue) shouldBe
+      Some(KubernetesConfig.jupyterTexeraOrigin)
+  }
+
+  "basePathFor" should "put the uid in the path so the gateway can read it" in {
+    // The browser cannot present Texera credentials on the requests Jupyter's own scripts
+    // make, so the owner has to be recoverable from the URL alone.
+    bare.basePathFor(7) shouldBe "/jupyter/7"
+  }
+
+  it should "give every user a distinct base path" in {
+    (1 to 50).map(bare.basePathFor).distinct.size shouldBe 50
+  }
+
+  it should "carry the configured image, pull policy and port" in {
+    val container = createdPod(7, "tok").getSpec.getContainers.asScala.head
+    container.getImage shouldBe KubernetesConfig.jupyterImageName
+    container.getImagePullPolicy shouldBe KubernetesConfig.computingUnitImagePullPolicy
+    container.getPorts.asScala.map(_.getContainerPort.intValue()) should contain(
+      KubernetesConfig.jupyterPortNumber
+    )
+  }
+
+  it should "carry the configured cpu and memory limits" in {
+    val limits = createdPod(7, "tok").getSpec.getContainers.asScala.head.getResources.getLimits
+    limits.get("cpu").toString shouldBe KubernetesConfig.jupyterCpuLimit
+    limits.get("memory").toString shouldBe KubernetesConfig.jupyterMemoryLimit
+  }
+
+  it should "set hostname and subdomain so generatePodURI resolves" in {
+    // The pair is what makes <pod>.<service>.<namespace>.svc.cluster.local addressable.
+    val spec = createdPod(7, "tok").getSpec
+    spec.getHostname shouldBe "jupyter-7"
+    spec.getSubdomain shouldBe KubernetesConfig.jupyterServiceName
+  }
+
+  "inCluster" should "build a client lazily without requiring a reachable cluster" in {
+    // The companion is only touched when a provision happens, but building the client must
+    // not itself need a cluster: single-node and local dev have none.
+    val client = JupyterKubernetesClient.inCluster
+    client.generatePodName(7) shouldBe "jupyter-7"
+  }
+
+  it should "create the pod in the Jupyter namespace" in {
+    val client = mock(classOf[Fabric8Client])
+    val namespaceable = mock(classOf[NamespaceableResource[_]])
+      .asInstanceOf[NamespaceableResource[Pod]]
+    val resource = mock(classOf[Resource[_]]).asInstanceOf[Resource[Pod]]
+    when(client.resource(org.mockito.ArgumentMatchers.any(classOf[Pod]))).thenReturn(namespaceable)
+    when(namespaceable.inNamespace(namespace)).thenReturn(resource)
+    new JupyterKubernetesClient(client).createPod(7, "tok")
+    verify(namespaceable).inNamespace(namespace)
+    verify(resource).create()
+  }
+}
diff --git a/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterProbeSpec.scala b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterProbeSpec.scala
new file mode 100644
index 00000000000..f6b39da35bc
--- /dev/null
+++ b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterProbeSpec.scala
@@ -0,0 +1,70 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import com.sun.net.httpserver.{HttpExchange, HttpServer}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.net.InetSocketAddress
+
+class JupyterProbeSpec extends AnyFlatSpec with Matchers {
+
+  // Binds an ephemeral port, so this never collides with the fixed-port stub in
+  // NotebookMigrationResourceSpec.
+  private def withServer(status: Int)(test: String => Unit): Unit = {
+    val server = HttpServer.create(new InetSocketAddress("localhost", 0), 0)
+    server.createContext(
+      "/api",
+      (exchange: HttpExchange) => {
+        exchange.getRequestBody.readAllBytes()
+        val body = """{"version":"2.7.0"}""".getBytes("UTF-8")
+        exchange.sendResponseHeaders(status, body.length)
+        val os = exchange.getResponseBody
+        os.write(body)
+        os.close()
+      }
+    )
+    server.start()
+    try test(s"http://localhost:${server.getAddress.getPort}")
+    finally server.stop(0)
+  }
+
+  "isAvailable" should "treat 200 as reachable" in {
+    withServer(200)(JupyterProbe.isAvailable(_) shouldBe true)
+  }
+
+  it should "treat 403 as reachable" in {
+    // /api needs no token, so a refusal still proves the server is up.
+    withServer(403)(JupyterProbe.isAvailable(_) shouldBe true)
+  }
+
+  it should "treat any other status as unavailable" in {
+    withServer(500)(JupyterProbe.isAvailable(_) shouldBe false)
+  }
+
+  it should "report unavailable when nothing is listening" in {
+    // Port 1 is reserved and unbound; the connect fails rather than hanging.
+    JupyterProbe.isAvailable("http://localhost:1") shouldBe false
+  }
+
+  it should "report unavailable for a malformed URL without opening a connection" in {
+    // The URL itself throws, so the cleanup path runs with no connection to close.
+    JupyterProbe.isAvailable("notaprotocol://host") shouldBe false
+  }
+}
diff --git a/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterTokenDeriverSpec.scala b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterTokenDeriverSpec.scala
new file mode 100644
index 00000000000..d1fcebafdc4
--- /dev/null
+++ b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterTokenDeriverSpec.scala
@@ -0,0 +1,82 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.texera.service.util
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class JupyterTokenDeriverSpec extends AnyFlatSpec with Matchers {
+
+  private val secret = "test-secret"
+
+  "JupyterTokenDeriver.derive" should "return the same token for a uid across calls" in {
+    // The token is never stored, so every call has to reproduce it or a running pod
+    // becomes unreachable.
+    JupyterTokenDeriver.derive(7, secret) shouldBe JupyterTokenDeriver.derive(7, secret)
+  }
+
+  it should "return a different token for each uid" in {
+    // The isolation property: one user's token must not open another user's Jupyter.
+    val tokens = (1 to 50).map(JupyterTokenDeriver.derive(_, secret))
+    tokens.distinct.size shouldBe 50
+  }
+
+  it should "return a different token when the secret changes" in {
+    // Rotation: changing the secret has to invalidate previously issued tokens.
+    JupyterTokenDeriver.derive(7, secret) should not be JupyterTokenDeriver.derive(7, "other")
+  }
+
+  it should "return a fixed-length lowercase hex token" in {
+    JupyterTokenDeriver.derive(7, secret) should fullyMatch regex "[0-9a-f]{32}"
+  }
+
+  it should "reject an empty secret" in {
+    an[IllegalArgumentException] should be thrownBy JupyterTokenDeriver.derive(7, "")
+  }
+
+  it should "fall back to the configured secret when none is passed" in {
+    // Exercises the default argument. The configured secret is empty unless a deployment
+    // sets one, so there is nothing to derive from and the require fires.
+    if (sys.env.get("JUPYTER_TOKEN_SECRET").isEmpty) {
+      an[IllegalArgumentException] should be thrownBy JupyterTokenDeriver.derive(7)
+    }
+  }
+
+  "JupyterTokenDeriver.validateConfiguration" should "reject an empty secret when per-user Jupyter is on" in {
+    val thrown = the[IllegalStateException] thrownBy JupyterTokenDeriver.validateConfiguration(
+      jupyterEnabled = true,
+      secret = ""
+    )
+    thrown.getMessage should include("storage.jupyter.token-secret")
+  }
+
+  it should "allow an empty secret when per-user Jupyter is off" in {
+    // Single-node and local dev run one shared Jupyter from static config and set no secret.
+    noException should be thrownBy JupyterTokenDeriver.validateConfiguration(
+      jupyterEnabled = false,
+      secret = ""
+    )
+  }
+
+  it should "allow a configured secret when per-user Jupyter is on" in {
+    noException should be thrownBy JupyterTokenDeriver.validateConfiguration(
+      jupyterEnabled = true,
+      secret = secret
+    )
+  }
+}
diff --git a/sql/changelog.xml b/sql/changelog.xml
index b57ec95ca42..5f0510e8003 100644
--- a/sql/changelog.xml
+++ b/sql/changelog.xml
@@ -124,6 +124,11 @@
         <sqlFile path="sql/updates/42.sql"/>
     </changeSet>
 
+    <!-- Add the per-user JupyterLab registry table -->
+    <changeSet id="43" author="zyratlo">
+        <sqlFile path="sql/updates/43.sql"/>
+    </changeSet>
+
     <!-- example changeSet
     <changeSet id="1" author="author">
         <sqlFile path="sql/updates/1.sql"/>
diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql
index acd4eef8177..f3622dc506e 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -83,6 +83,7 @@ DROP TABLE IF EXISTS site_settings CASCADE;
 DROP TABLE IF EXISTS computing_unit_user_access CASCADE;
 DROP TABLE IF EXISTS notebook CASCADE;
 DROP TABLE IF EXISTS workflow_notebook_mapping CASCADE;
+DROP TABLE IF EXISTS user_jupyter CASCADE;
 DROP TABLE IF EXISTS virtual_environments CASCADE;
 
 -- ============================================
@@ -670,6 +671,20 @@ CREATE TABLE IF NOT EXISTS workflow_notebook_mapping
     FOREIGN KEY (wid, nid) REFERENCES notebook(wid, nid) ON DELETE CASCADE
 );
 
+-- Per-user JupyterLab registration: one row per provisioned user, so the service resolves
+-- endpoints per user instead of from process-wide config. Both URLs are stored rather than
+-- derived, matching workflow_computing_unit.uri, so addressing can change without a code
+-- change: internal_url is what the service calls, public_url is what the browser loads.
+-- The token is derived from a server secret and the uid, so it is not stored.
+CREATE TABLE IF NOT EXISTS user_jupyter
+(
+    uid          INT          NOT NULL PRIMARY KEY,
+    internal_url VARCHAR(512) NOT NULL,
+    public_url   VARCHAR(512) NOT NULL,
+    created_at   TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    FOREIGN KEY (uid) REFERENCES "user"(uid) ON DELETE CASCADE
+);
+
 -- START Fulltext search index creation (DO NOT EDIT THIS LINE)
 CREATE EXTENSION IF NOT EXISTS pgroonga;
 
diff --git a/sql/updates/43.sql b/sql/updates/43.sql
new file mode 100644
index 00000000000..c00d7d6d01d
--- /dev/null
+++ b/sql/updates/43.sql
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+\c texera_db
+
+SET search_path TO texera_db;
+
+BEGIN;
+
+-- Per-user JupyterLab registration: one row per provisioned user, so the service resolves
+-- endpoints per user instead of from process-wide config. Both URLs are stored rather than
+-- derived, matching workflow_computing_unit.uri, so addressing can change without a code
+-- change: internal_url is what the service calls, public_url is what the browser loads.
+-- The token is derived from a server secret and the uid, so it is not stored.
+CREATE TABLE IF NOT EXISTS user_jupyter
+(
+    uid          INT          NOT NULL PRIMARY KEY,
+    internal_url VARCHAR(512) NOT NULL,
+    public_url   VARCHAR(512) NOT NULL,
+    created_at   TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    FOREIGN KEY (uid) REFERENCES "user"(uid) ON DELETE CASCADE
+);
+
+COMMIT;