From 8c49cf38dbe6024096d929505011e9be71989062 Mon Sep 17 00:00:00 2001
From: Ryan Zhang <zyratlo@gmail.com>
Date: Wed, 26 Aug 2026 19:18:54 -0700
Subject: [PATCH 1/8] feat(dao): add the per-user Jupyter registry table

---
 sql/changelog.xml  |  5 +++++
 sql/texera_ddl.sql | 15 +++++++++++++++
 sql/updates/42.sql | 40 ++++++++++++++++++++++++++++++++++++++++
 3 files changed, 60 insertions(+)
 create mode 100644 sql/updates/42.sql

diff --git a/sql/changelog.xml b/sql/changelog.xml
index e86f869665a..38ac4db0a2b 100644
--- a/sql/changelog.xml
+++ b/sql/changelog.xml
@@ -119,6 +119,11 @@
         <sqlFile path="sql/updates/41.sql"/>
     </changeSet>
 
+    <!-- Add the per-user JupyterLab registry table -->
+    <changeSet id="42" author="zyratlo">
+        <sqlFile path="sql/updates/42.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/42.sql b/sql/updates/42.sql
new file mode 100644
index 00000000000..c00d7d6d01d
--- /dev/null
+++ b/sql/updates/42.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;

From 29ea5297ec380c674e7d08d8d87c7bf115953f5a Mon Sep 17 00:00:00 2001
From: Ryan Zhang <zyratlo@gmail.com>
Date: Wed, 26 Aug 2026 19:27:43 -0700
Subject: [PATCH 2/8] feat(config): add per-user Jupyter provisioning settings

---
 .../config/src/main/resources/kubernetes.conf | 29 +++++++++++++++++++
 common/config/src/main/resources/storage.conf |  6 ++++
 .../common/config/KubernetesConfig.scala      | 13 +++++++++
 .../texera/common/config/StorageConfig.scala  |  3 ++
 .../common/config/KubernetesConfigSpec.scala  | 27 ++++++++++++++++-
 .../common/config/StorageConfigSpec.scala     |  8 +++++
 6 files changed, 85 insertions(+), 1 deletion(-)

diff --git a/common/config/src/main/resources/kubernetes.conf b/common/config/src/main/resources/kubernetes.conf
index e85924e570c..8c7f7521b84 100644
--- a/common/config/src/main/resources/kubernetes.conf
+++ b/common/config/src/main/resources/kubernetes.conf
@@ -54,4 +54,33 @@ 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
+
+  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..59d6f9a6dd0 100644
--- a/common/config/src/main/resources/storage.conf
+++ b/common/config/src/main/resources/storage.conf
@@ -188,5 +188,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..7e8ebd991d7 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,17 @@ 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 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..466cc06b860 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,31 @@ 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"
+    )
+    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 ""
+    }
+  }
 }

From 5fccbc3972cd61d3a43df3091974cd0ebb9ba97f Mon Sep 17 00:00:00 2001
From: Ryan Zhang <zyratlo@gmail.com>
Date: Wed, 26 Aug 2026 19:33:31 -0700
Subject: [PATCH 3/8] feat(notebook-migration): derive per-user Jupyter tokens

---
 .../service/NotebookMigrationService.scala    |  4 +
 .../service/util/JupyterTokenDeriver.scala    | 59 +++++++++++++++
 .../util/JupyterTokenDeriverSpec.scala        | 74 +++++++++++++++++++
 3 files changed, 137 insertions(+)
 create mode 100644 notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterTokenDeriver.scala
 create mode 100644 notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterTokenDeriverSpec.scala

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/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/util/JupyterTokenDeriverSpec.scala b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterTokenDeriverSpec.scala
new file mode 100644
index 00000000000..7be1e048a72
--- /dev/null
+++ b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterTokenDeriverSpec.scala
@@ -0,0 +1,74 @@
+// 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, "")
+  }
+
+  "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
+    )
+  }
+}

From 7b27436a019adeb661d8e2ee2ca9264e39f96e20 Mon Sep 17 00:00:00 2001
From: Ryan Zhang <zyratlo@gmail.com>
Date: Wed, 26 Aug 2026 19:41:44 -0700
Subject: [PATCH 4/8] feat(notebook-migration): resolve Jupyter endpoints per
 user

---
 .../resource/NotebookMigrationResource.scala  | 50 ++++++-----
 .../util/JupyterEndpointResolver.scala        | 61 +++++++++++++
 .../service/util/JupyterEndpoints.scala       | 37 ++++++++
 .../NotebookMigrationResourceSpec.scala       | 87 ++++++++++++++++++-
 4 files changed, 209 insertions(+), 26 deletions(-)
 create mode 100644 notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterEndpointResolver.scala
 create mode 100644 notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterEndpoints.scala

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..2be513b06be 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,7 @@ 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}
 
 object NotebookMigrationResource extends LazyLogging {
 
@@ -106,21 +106,6 @@ 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"
@@ -149,7 +134,7 @@ object NotebookMigrationResource extends LazyLogging {
   // 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.
@@ -172,7 +157,7 @@ object NotebookMigrationResource extends LazyLogging {
   }
 
   // Returns the URL of Jupyter
-  def getJupyterURL(jupyter: JupyterEndpoints = configuredEndpoints): Response = {
+  def getJupyterURL(jupyter: JupyterEndpoints = JupyterEndpoints.configured): Response = {
     if (!isJupyterAvailable(jupyter.internalUrl)) {
       return jupyterUnavailableResponse
     }
@@ -181,7 +166,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 {
@@ -271,7 +259,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 {
@@ -566,6 +557,15 @@ object NotebookMigrationResource extends LazyLogging {
 @Consumes(Array(MediaType.APPLICATION_JSON))
 class NotebookMigrationResource extends LazyLogging {
 
+  // Runs `call` against the caller's own Jupyter. The uid comes from the authenticated
+  // session, so a request cannot name another user's server. None means the user has no
+  // Jupyter provisioned, which is reported the same as an unreachable one.
+  private def withJupyter(user: SessionUser)(call: JupyterEndpoints => Response): Response =
+    JupyterEndpointResolver.resolve(user.getUid) match {
+      case Some(jupyter) => call(jupyter)
+      case None          => NotebookMigrationResource.jupyterUnavailableResponse
+    }
+
   @GET
   @Path("/get-jupyter-iframe-url")
   def getJupyterIframeURL(
@@ -576,28 +576,30 @@ class NotebookMigrationResource extends LazyLogging {
     val name = Option(notebookName)
       .filter(_.nonEmpty)
       .getOrElse(NotebookMigrationResource.defaultNotebookName)
-    NotebookMigrationResource.getJupyterIframeURL(name)
+    withJupyter(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)
+    withJupyter(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/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..e46a18780d9 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,29 @@ 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,
+  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
@@ -152,6 +160,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 +552,86 @@ 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 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 +676,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"

From c2f6aabc73abc5913924a436e4e4f4f7cb448ed1 Mon Sep 17 00:00:00 2001
From: Ryan Zhang <zyratlo@gmail.com>
Date: Wed, 26 Aug 2026 20:00:43 -0700
Subject: [PATCH 5/8] feat(notebook-migration): provision per-user JupyterLab
 pods

---
 notebook-migration-service/LICENSE-binary     |  36 ++++-
 notebook-migration-service/build.sbt          |   3 +-
 .../resource/NotebookMigrationResource.scala  |  62 ++++----
 .../util/JupyterKubernetesClient.scala        | 105 +++++++++++++
 .../texera/service/util/JupyterProbe.scala    |  46 ++++++
 .../service/util/JupyterProvisioner.scala     | 143 ++++++++++++++++++
 .../NotebookMigrationResourceSpec.scala       | 118 +++++++++++++++
 .../util/JupyterKubernetesClientSpec.scala    |  50 ++++++
 8 files changed, 525 insertions(+), 38 deletions(-)
 create mode 100644 notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterKubernetesClient.scala
 create mode 100644 notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterProbe.scala
 create mode 100644 notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterProvisioner.scala
 create mode 100644 notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterKubernetesClientSpec.scala

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/scala/org/apache/texera/service/resource/NotebookMigrationResource.scala b/notebook-migration-service/src/main/scala/org/apache/texera/service/resource/NotebookMigrationResource.scala
index 2be513b06be..9debd18705d 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.service.util.{JupyterEndpointResolver, JupyterEndpoints}
+import org.apache.texera.service.util.{
+  JupyterEndpointResolver,
+  JupyterEndpoints,
+  JupyterProbe,
+  JupyterProvisioner
+}
 
 object NotebookMigrationResource extends LazyLogging {
 
@@ -110,27 +115,6 @@ object NotebookMigrationResource extends LazyLogging {
   // 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,
@@ -145,7 +129,7 @@ object NotebookMigrationResource extends LazyLogging {
         .build()
     }
 
-    if (!isJupyterAvailable(jupyter.internalUrl)) {
+    if (!JupyterProbe.isAvailable(jupyter.internalUrl)) {
       return jupyterUnavailableResponse
     }
 
@@ -158,7 +142,7 @@ object NotebookMigrationResource extends LazyLogging {
 
   // Returns the URL of Jupyter
   def getJupyterURL(jupyter: JupyterEndpoints = JupyterEndpoints.configured): Response = {
-    if (!isJupyterAvailable(jupyter.internalUrl)) {
+    if (!JupyterProbe.isAvailable(jupyter.internalUrl)) {
       return jupyterUnavailableResponse
     }
 
@@ -190,7 +174,7 @@ object NotebookMigrationResource extends LazyLogging {
           .build()
       }
 
-      if (!isJupyterAvailable(jupyter.internalUrl)) {
+      if (!JupyterProbe.isAvailable(jupyter.internalUrl)) {
         return jupyterUnavailableResponse
       }
 
@@ -281,7 +265,7 @@ object NotebookMigrationResource extends LazyLogging {
           .build()
       }
 
-      if (!isJupyterAvailable(jupyter.internalUrl)) {
+      if (!JupyterProbe.isAvailable(jupyter.internalUrl)) {
         return jupyterUnavailableResponse
       }
 
@@ -557,13 +541,23 @@ object NotebookMigrationResource extends LazyLogging {
 @Consumes(Array(MediaType.APPLICATION_JSON))
 class NotebookMigrationResource extends LazyLogging {
 
-  // Runs `call` against the caller's own Jupyter. The uid comes from the authenticated
-  // session, so a request cannot name another user's server. None means the user has no
-  // Jupyter provisioned, which is reported the same as an unreachable one.
+  // 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 =
-    JupyterEndpointResolver.resolve(user.getUid) match {
-      case Some(jupyter) => call(jupyter)
-      case None          => NotebookMigrationResource.jupyterUnavailableResponse
+    respondWith(JupyterEndpointResolver.resolve(user.getUid), call)
+
+  private def respondWith(
+      jupyter: Option[JupyterEndpoints],
+      call: JupyterEndpoints => Response
+  ): Response =
+    jupyter match {
+      case Some(endpoints) => call(endpoints)
+      case None            => NotebookMigrationResource.jupyterUnavailableResponse
     }
 
   @GET
@@ -576,7 +570,7 @@ class NotebookMigrationResource extends LazyLogging {
     val name = Option(notebookName)
       .filter(_.nonEmpty)
       .getOrElse(NotebookMigrationResource.defaultNotebookName)
-    withJupyter(user) { jupyter =>
+    withNewJupyter(user) { jupyter =>
       NotebookMigrationResource.getJupyterIframeURL(name, jupyter)
     }
   }
@@ -592,7 +586,7 @@ class NotebookMigrationResource extends LazyLogging {
   @Path("/set-notebook")
   def setNotebook(body: String, @Auth user: SessionUser): Response = {
     logger.info("Setting notebook")
-    withJupyter(user)(NotebookMigrationResource.setNotebook(body, _))
+    withNewJupyter(user)(NotebookMigrationResource.setNotebook(body, _))
   }
 
   @POST
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..3fc50feed5a
--- /dev/null
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterKubernetesClient.scala
@@ -0,0 +1,105 @@
+// 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}"
+
+  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()
+      )
+      .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..e2ac37525de
--- /dev/null
+++ b/notebook-migration-service/src/main/scala/org/apache/texera/service/util/JupyterProvisioner.scala
@@ -0,0 +1,143 @@
+// 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] = {
+    val internalUrl = s"http://${kubernetes.generatePodURI(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/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala b/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala
index e46a18780d9..be1feb3a5cd 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
@@ -25,6 +25,8 @@ 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}
@@ -57,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
 
@@ -632,6 +636,120 @@ class NotebookMigrationResourceSpec
     writerToken should not be readerToken
   }
 
+  // -- 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
+    override def podExists(uid: Int): Boolean = alreadyExists
+    override def createPod(uid: Int, token: String) = { created ::= ((uid, token)); null }
+    override def deletePod(uid: Int): Unit = deleted ::= uid
+  }
+
+  // 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)}")
+  }
+
+  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)}")
+  }
+
+  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 "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 "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)
+  }
+
   "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)
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..9206c2c3b8d
--- /dev/null
+++ b/notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterKubernetesClientSpec.scala
@@ -0,0 +1,50 @@
+// 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
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+  * Covers the pure naming and addressing, which is what the registry stores and the service
+  * dials. The fabric8-backed calls need a cluster and are left to deployment.
+  */
+class JupyterKubernetesClientSpec extends AnyFlatSpec with Matchers {
+
+  private val client = new JupyterKubernetesClient(null)
+
+  "generatePodName" should "namespace the pod by uid" in {
+    client.generatePodName(7) shouldBe "jupyter-7"
+  }
+
+  it should "give every user a distinct pod name" in {
+    (1 to 50).map(client.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.
+    client.generatePodURI(7) shouldBe
+      s"jupyter-7.${KubernetesConfig.jupyterServiceName}.${KubernetesConfig.jupyterNamespace}" +
+        s".svc.cluster.local:${KubernetesConfig.jupyterPortNumber}"
+  }
+
+  it should "carry the configured port" in {
+    client.generatePodURI(7) should endWith(s":${KubernetesConfig.jupyterPortNumber}")
+  }
+}

From dbbba4ffac418d0b27532932c4f2a3ca7f69c59a Mon Sep 17 00:00:00 2001
From: Ryan Zhang <zyratlo@gmail.com>
Date: Wed, 26 Aug 2026 20:04:13 -0700
Subject: [PATCH 6/8] docs(notebook-migration): note the static Jupyter is the
 fallback

---
 common/config/src/main/resources/storage.conf                  | 3 ++-
 .../texera/service/resource/NotebookMigrationResource.scala    | 2 ++
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/common/config/src/main/resources/storage.conf b/common/config/src/main/resources/storage.conf
index 59d6f9a6dd0..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}
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 9debd18705d..557a1638737 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
@@ -63,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)

From 621f1fff38001bdc484f204684b76f43d4a612e0 Mon Sep 17 00:00:00 2001
From: Ryan Zhang <zyratlo@gmail.com>
Date: Thu, 27 Aug 2026 13:14:29 -0700
Subject: [PATCH 7/8] test(notebook-migration): cover per-user Jupyter
 provisioning paths

---
 .../resource/NotebookMigrationResource.scala  |   4 +-
 .../NotebookMigrationResourceSpec.scala       |  95 ++++++++++-
 .../service/util/JupyterEndpointsSpec.scala   |  60 +++++++
 .../util/JupyterKubernetesClientSpec.scala    | 154 +++++++++++++++++-
 .../service/util/JupyterProbeSpec.scala       |  70 ++++++++
 .../util/JupyterTokenDeriverSpec.scala        |   8 +
 6 files changed, 380 insertions(+), 11 deletions(-)
 create mode 100644 notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterEndpointsSpec.scala
 create mode 100644 notebook-migration-service/src/test/scala/org/apache/texera/service/util/JupyterProbeSpec.scala

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 557a1638737..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
@@ -553,7 +553,9 @@ class NotebookMigrationResource extends LazyLogging {
   private def withJupyter(user: SessionUser)(call: JupyterEndpoints => Response): Response =
     respondWith(JupyterEndpointResolver.resolve(user.getUid), call)
 
-  private def respondWith(
+  // 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 =
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 be1feb3a5cd..e446faa9190 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
@@ -636,6 +636,20 @@ class NotebookMigrationResourceSpec
     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
@@ -644,9 +658,18 @@ class NotebookMigrationResourceSpec
     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) = { created ::= ((uid, token)); null }
-    override def deletePod(uid: Int): Unit = deleted ::= uid
+    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.
@@ -736,6 +759,64 @@ class NotebookMigrationResourceSpec
     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)}")
+  }
+
+  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.
@@ -750,6 +831,16 @@ class NotebookMigrationResourceSpec
     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)
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
index 9206c2c3b8d..6c3f4faf4d9 100644
--- 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
@@ -17,34 +17,172 @@
 
 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._
+
 /**
-  * Covers the pure naming and addressing, which is what the registry stores and the service
-  * dials. The fabric8-backed calls need a cluster and are left to deployment.
+  * 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 client = new JupyterKubernetesClient(null)
+  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 {
-    client.generatePodName(7) shouldBe "jupyter-7"
+    bare.generatePodName(7) shouldBe "jupyter-7"
   }
 
   it should "give every user a distinct pod name" in {
-    (1 to 50).map(client.generatePodName).distinct.size shouldBe 50
+    (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.
-    client.generatePodURI(7) shouldBe
-      s"jupyter-7.${KubernetesConfig.jupyterServiceName}.${KubernetesConfig.jupyterNamespace}" +
+    bare.generatePodURI(7) shouldBe
+      s"jupyter-7.${KubernetesConfig.jupyterServiceName}.$namespace" +
         s".svc.cluster.local:${KubernetesConfig.jupyterPortNumber}"
   }
 
   it should "carry the configured port" in {
-    client.generatePodURI(7) should endWith(s":${KubernetesConfig.jupyterPortNumber}")
+    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 "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
index 7be1e048a72..d1fcebafdc4 100644
--- 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
@@ -49,6 +49,14 @@ class JupyterTokenDeriverSpec extends AnyFlatSpec with Matchers {
     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,

From 06546e27fc259b6d15f9a60c0fc4378813c05339 Mon Sep 17 00:00:00 2001
From: Ryan Zhang <zyratlo@gmail.com>
Date: Fri, 28 Aug 2026 11:50:26 -0700
Subject: [PATCH 8/8] fix(dao): renumber the Jupyter registry migration after
 42 was taken

---
 sql/changelog.xml              | 4 ++--
 sql/updates/{42.sql => 43.sql} | 0
 2 files changed, 2 insertions(+), 2 deletions(-)
 rename sql/updates/{42.sql => 43.sql} (100%)

diff --git a/sql/changelog.xml b/sql/changelog.xml
index 38ac4db0a2b..7f7b693d63d 100644
--- a/sql/changelog.xml
+++ b/sql/changelog.xml
@@ -120,8 +120,8 @@
     </changeSet>
 
     <!-- Add the per-user JupyterLab registry table -->
-    <changeSet id="42" author="zyratlo">
-        <sqlFile path="sql/updates/42.sql"/>
+    <changeSet id="43" author="zyratlo">
+        <sqlFile path="sql/updates/43.sql"/>
     </changeSet>
 
     <!-- example changeSet
diff --git a/sql/updates/42.sql b/sql/updates/43.sql
similarity index 100%
rename from sql/updates/42.sql
rename to sql/updates/43.sql