diff --git a/.dockerignore b/.dockerignore
deleted file mode 100644
index ab25602c..00000000
--- a/.dockerignore
+++ /dev/null
@@ -1,5 +0,0 @@
-*~
-.*.sw[nop]
-.DS_Store
-.idea
-/.git/
diff --git a/.env.example b/.env.example
index 26e6520f..3174018a 100644
--- a/.env.example
+++ b/.env.example
@@ -56,7 +56,13 @@ GOOGLE_APPLICATION_CREDENTIALS="/path/to/your/service-account-key.json"
 # Custom Dataproc API endpoint (uncomment if needed)
 # GOOGLE_CLOUD_DATAPROC_API_ENDPOINT="your-region-dataproc.googleapis.com"
 
-# Subnet URI for Managed Spark Connect (full resource name format)
-# Example: projects/your-project-id/regions/us-central1/subnetworks/your-subnet-name
+# Subnet name or URI for Managed Spark Connect (optional; defaults to "default" in tests)
+# Accepts a short subnet name (e.g. "default") or full resource URI:
+# MANAGED_SPARK_CONNECT_SUBNET="default"
 # MANAGED_SPARK_CONNECT_SUBNET="projects/your-project-id/regions/us-central1/subnetworks/your-subnet-name"
 
+# Comma-separated fallback regions for integration tests when a short subnet name is used
+# and the primary region encounters a 503 stockout (defaults to us-central1,us-east1,us-west1)
+# GOOGLE_CLOUD_FALLBACK_REGIONS="us-central1,us-east1,us-west1"
+
+
diff --git a/.kokoro/trampoline.sh b/.github/dependabot.yml
old mode 100755
new mode 100644
similarity index 51%
rename from .kokoro/trampoline.sh
rename to .github/dependabot.yml
index f2956923..2c1e3718
--- a/.kokoro/trampoline.sh
+++ b/.github/dependabot.yml
@@ -1,11 +1,10 @@
-#!/bin/bash
-# Copyright 2024 Google LLC
+# Copyright 2026 Google LLC
 #
 # Licensed 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
+#     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,
@@ -13,16 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-set -eo pipefail
-
-# Always run the cleanup script, regardless of the success of bouncing into
-# the container.
-function cleanup() {
-    chmod +x ${KOKORO_GFILE_DIR}/trampoline_cleanup.sh
-    ${KOKORO_GFILE_DIR}/trampoline_cleanup.sh
-    echo "cleanup";
-}
-trap cleanup EXIT
-
-$(dirname $0)/populate-secrets.sh # Secret Manager secrets.
-python3 "${KOKORO_GFILE_DIR}/trampoline_v1.py"
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"
+    groups:
+      github-actions:
+        patterns:
+          - "*"
diff --git a/.github/workflows/integration-tests.yaml b/.github/workflows/integration-tests.yaml
index 3b0dadec..a410bec7 100644
--- a/.github/workflows/integration-tests.yaml
+++ b/.github/workflows/integration-tests.yaml
@@ -16,34 +16,105 @@
 #
 # Required GitHub Secrets:
 #   - GCP_SA_KEY: Service account JSON key (project_id and client_email extracted automatically)
-#   - GCP_REGION: Google Cloud Region (optional, defaults to us-central1)
-#   - GCP_SUBNET: Managed Spark subnet name or URI (optional, defaults to "default")
 #
-# See INTEGRATION_TESTS.md for setup instructions.
+# Optional GitHub Repository Variables (vars.*):
+#   - GCP_REGION: Google Cloud Region (defaults to us-central1)
+#   - GCP_SUBNET: Managed Spark subnet name or URI (defaults to "default")
+#
+# Fork PRs can be tested by a repository maintainer commenting `/gcbrun` on the PR.
+# See DEVELOPING.md for local setup instructions.
 
 name: Integration Tests
 on:
+  push:
+    branches: [ main ]
   pull_request:
     branches: [ main ]
+  issue_comment:
+    types: [ created ]
   workflow_dispatch:
 
 permissions:
   contents: read
+  pull-requests: read
+  statuses: write
 
 jobs:
   integration-test:
     name: Run integration tests
     runs-on: ubuntu-latest
+    timeout-minutes: 25
+
+    # Run on push to main, manual dispatch, non-Dependabot same-repo pull requests,
+    # or maintainer /gcbrun comments on PRs.
+    if: >-
+      github.event_name == 'push' ||
+      github.event_name == 'workflow_dispatch' ||
+      (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == github.repository && github.actor != 'dependabot[bot]') ||
+      (github.event_name == 'issue_comment' && github.event.issue.pull_request != null && contains(fromJSON('["OWNER", "MEMBER", "COLLABORATOR"]'), github.event.comment.author_association) && (github.event.comment.body == '/gcbrun' || startsWith(github.event.comment.body, '/gcbrun ') || startsWith(github.event.comment.body, '/gcbrun\r') || startsWith(github.event.comment.body, '/gcbrun\n')))
 
-    # Only run integration tests if secrets are available
-    if: github.event_name == 'workflow_dispatch' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == github.repository)
+    concurrency:
+      group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.event.issue.number || github.ref }}
+      cancel-in-progress: ${{ github.event_name == 'pull_request' || github.event_name == 'issue_comment' }}
 
     steps:
+    - name: Authorize /gcbrun and resolve PR head SHA
+      if: github.event_name == 'issue_comment'
+      env:
+        GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        COMMENT_AUTHOR: ${{ github.event.comment.user.login }}
+        COMMENT_BODY: ${{ github.event.comment.body }}
+        COMMENT_CREATED_AT: ${{ github.event.comment.created_at }}
+        PR_NUMBER: ${{ github.event.issue.number }}
+        RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+      run: |
+        PERMISSION=$(gh api "repos/${GITHUB_REPOSITORY}/collaborators/${COMMENT_AUTHOR}/permission" --jq '.permission')
+        if [[ "$PERMISSION" != "admin" && "$PERMISSION" != "maintain" && "$PERMISSION" != "write" ]]; then
+          echo "::error::User ${COMMENT_AUTHOR} (permission: ${PERMISSION}) is not authorized to trigger /gcbrun."
+          exit 1
+        fi
+        PR_JSON=$(gh api "repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}")
+        PR_STATE=$(echo "$PR_JSON" | jq -r '.state')
+        if [[ "$PR_STATE" != "open" ]]; then
+          echo "::error::PR #${PR_NUMBER} is not open (state: ${PR_STATE})."
+          exit 1
+        fi
+        PR_HEAD_SHA=$(echo "$PR_JSON" | jq -r '.head.sha')
+        REQUESTED_SHA=$(echo "$COMMENT_BODY" | head -n1 | tr -d '\r' | sed -E 's|^/gcbrun[[:space:]]*||; s|[[:space:]]+$||')
+        if [[ -n "$REQUESTED_SHA" && "$PR_HEAD_SHA" != "$REQUESTED_SHA"* ]]; then
+          echo "::error::PR head (${PR_HEAD_SHA}) does not match requested /gcbrun SHA (${REQUESTED_SHA})."
+          exit 1
+        fi
+        SUITE_CREATED_AT=$(gh api "repos/${GITHUB_REPOSITORY}/commits/${PR_HEAD_SHA}/check-suites" --jq '[.check_suites[].created_at] | min // empty' || true)
+        if [[ -n "$SUITE_CREATED_AT" && "$SUITE_CREATED_AT" > "$COMMENT_CREATED_AT" ]]; then
+          echo "::error::PR head commit ${PR_HEAD_SHA} was pushed at ${SUITE_CREATED_AT}, after /gcbrun comment at ${COMMENT_CREATED_AT}."
+          exit 1
+        fi
+        echo "PR_HEAD_SHA=${PR_HEAD_SHA}" >> "$GITHUB_ENV"
+        gh api "repos/${GITHUB_REPOSITORY}/statuses/${PR_HEAD_SHA}" \
+          -f state="pending" \
+          -f target_url="${RUN_URL}" \
+          -f description="Integration tests running via /gcbrun" \
+          -f context="Integration Tests / Run integration tests"
+
     - name: Checkout code
       uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       with:
         persist-credentials: false
 
+    - name: Checkout PR head commit for /gcbrun
+      if: github.event_name == 'issue_comment'
+      env:
+        PR_NUMBER: ${{ github.event.issue.number }}
+      run: |
+        git fetch origin "refs/pull/${PR_NUMBER}/head"
+        FETCH_SHA=$(git rev-parse FETCH_HEAD)
+        if [[ "$FETCH_SHA" != "$PR_HEAD_SHA" ]]; then
+          echo "::error::Fetched PR head (${FETCH_SHA}) does not match authorized SHA (${PR_HEAD_SHA}). Re-run /gcbrun after reviewing the new commit."
+          exit 1
+        fi
+        git checkout --detach "${PR_HEAD_SHA}"
+
     - name: Setup Python
       uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
       with:
@@ -86,12 +157,32 @@ jobs:
     - name: Run integration tests
       env:
         CI: "true"
-        # Extracted from service account JSON in the previous step
+        # Extracted from service account JSON in the previous step (kept secret)
         GOOGLE_CLOUD_PROJECT: ${{ env.PROJECT_ID }}
         MANAGED_SPARK_CONNECT_SERVICE_ACCOUNT: ${{ env.SA_EMAIL }}
-        # Infrastructure-specific secrets
-        GOOGLE_CLOUD_REGION: ${{ secrets.GCP_REGION || 'us-central1' }}
-        MANAGED_SPARK_CONNECT_SUBNET: ${{ secrets.GCP_SUBNET }}
+        # Non-secret infrastructure variables (falling back to secrets during transition)
+        GOOGLE_CLOUD_REGION: ${{ vars.GCP_REGION || secrets.GCP_REGION || 'us-central1' }}
+        MANAGED_SPARK_CONNECT_SUBNET: ${{ vars.GCP_SUBNET || secrets.GCP_SUBNET }}
         MANAGED_SPARK_CONNECT_AUTH_TYPE: "SERVICE_ACCOUNT"
       run: |
-        python -m pytest tests/integration/ -v --tb=short -x
\ No newline at end of file
+        python -m pytest tests/integration/ -v --tb=short -x
+
+    - name: Report /gcbrun commit status
+      if: always() && github.event_name == 'issue_comment' && env.PR_HEAD_SHA != ''
+      env:
+        GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        JOB_STATUS: ${{ job.status }}
+        RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+      run: |
+        if [[ "$JOB_STATUS" == "success" ]]; then
+          STATE="success"
+          DESC="Integration tests passed via /gcbrun"
+        else
+          STATE="failure"
+          DESC="Integration tests ${JOB_STATUS} via /gcbrun"
+        fi
+        gh api "repos/${GITHUB_REPOSITORY}/statuses/${PR_HEAD_SHA}" \
+          -f state="${STATE}" \
+          -f target_url="${RUN_URL}" \
+          -f description="${DESC}" \
+          -f context="Integration Tests / Run integration tests"
diff --git a/.github/workflows/lint-readme.yaml b/.github/workflows/lint-readme.yaml
index 9f82131e..43900b29 100644
--- a/.github/workflows/lint-readme.yaml
+++ b/.github/workflows/lint-readme.yaml
@@ -21,12 +21,22 @@ on:
 permissions:
   contents: read
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
 jobs:
   lint-readme:
     name: lint-readme
     runs-on: ubuntu-latest
+    timeout-minutes: 5
 
     steps:
+    - name: Checkout code
+      uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      with:
+        persist-credentials: false
+
     - name: Setup Python
       uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
       with:
@@ -35,9 +45,6 @@ jobs:
     - name: Set up uv
       uses: astral-sh/setup-uv@1c37ad07a6a961277cf70c0d37d6f313000f5884 # v10.3.0
 
-    - name: Checkout code
-      uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
-
     - name: Install dependencies
       run: uv pip install --system build twine
 
diff --git a/.github/workflows/pyink.yaml b/.github/workflows/pyink.yaml
index 8159a776..4234d463 100644
--- a/.github/workflows/pyink.yaml
+++ b/.github/workflows/pyink.yaml
@@ -21,12 +21,22 @@ on:
 permissions:
   contents: read
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
 jobs:
   pyink:
     name: Run pyink
     runs-on: ubuntu-latest
+    timeout-minutes: 5
 
     steps:
+    - name: Checkout code
+      uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      with:
+        persist-credentials: false
+
     - name: Setup Python
       uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
       with:
@@ -36,10 +46,7 @@ jobs:
       uses: astral-sh/setup-uv@1c37ad07a6a961277cf70c0d37d6f313000f5884 # v10.3.0
 
     - name: Install pyink
-      run: uv pip install --system pyink
-
-    - name: Checkout code
-      uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      run: uv pip install --system 'pyink~=24.0'
 
     - name: Run pyink check
-      run: pyink . --check
\ No newline at end of file
+      run: pyink . --check
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index 9cb1561a..22a39574 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -14,6 +14,8 @@
 
 name: Unit Tests
 on:
+  push:
+    branches: [ main ]
   pull_request:
     branches: [ main ]
   workflow_dispatch:
@@ -21,10 +23,15 @@ on:
 permissions:
   contents: read
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
 jobs:
   test:
     name: Run unit tests ([${{ matrix.spark-extra }}])
     runs-on: ubuntu-latest
+    timeout-minutes: 10
     strategy:
       fail-fast: false
       matrix:
@@ -35,6 +42,8 @@ jobs:
     steps:
     - name: Checkout code
       uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      with:
+        persist-credentials: false
 
     - name: Setup Python
       uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
@@ -75,4 +84,4 @@ jobs:
       if: matrix.spark-extra == 'spark'
       run: python -m pytest tests/local/ -v --tb=short
       env:
-        JAVA_HOME: ${{ env.JAVA_HOME_17_X64 }}
\ No newline at end of file
+        JAVA_HOME: ${{ env.JAVA_HOME_17_X64 }}
diff --git a/.kokoro/populate-secrets.sh b/.kokoro/populate-secrets.sh
deleted file mode 100755
index c435402f..00000000
--- a/.kokoro/populate-secrets.sh
+++ /dev/null
@@ -1,43 +0,0 @@
-#!/bin/bash
-# Copyright 2024 Google LLC.
-#
-# Licensed 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.
-
-set -eo pipefail
-
-function now { date +"%Y-%m-%d %H:%M:%S" | tr -d '\n' ;}
-function msg { println "$*" >&2 ;}
-function println { printf '%s\n' "$(now) $*" ;}
-
-
-# Populates requested secrets set in SECRET_MANAGER_KEYS from service account:
-# kokoro-trampoline@cloud-devrel-kokoro-resources.iam.gserviceaccount.com
-SECRET_LOCATION="${KOKORO_GFILE_DIR}/secret_manager"
-msg "Creating folder on disk for secrets: ${SECRET_LOCATION}"
-mkdir -p ${SECRET_LOCATION}
-for key in $(echo ${SECRET_MANAGER_KEYS} | sed "s/,/ /g")
-do
-  msg "Retrieving secret ${key}"
-  docker run --entrypoint=gcloud \
-    --volume=${KOKORO_GFILE_DIR}:${KOKORO_GFILE_DIR} \
-    gcr.io/google.com/cloudsdktool/cloud-sdk \
-    secrets versions access latest \
-    --project cloud-devrel-kokoro-resources \
-    --secret ${key} > \
-    "${SECRET_LOCATION}/${key}"
-  if [[ $? == 0 ]]; then
-    msg "Secret written to ${SECRET_LOCATION}/${key}"
-  else
-    msg "Error retrieving secret ${key}"
-  fi
-done
diff --git a/.kokoro/trampoline_v2.sh b/.kokoro/trampoline_v2.sh
deleted file mode 100755
index bb6f5eda..00000000
--- a/.kokoro/trampoline_v2.sh
+++ /dev/null
@@ -1,487 +0,0 @@
-#!/usr/bin/env bash
-# Copyright 2024 Google LLC
-#
-# Licensed 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.
-
-# trampoline_v2.sh
-#
-# This script does 3 things.
-#
-# 1. Prepare the Docker image for the test
-# 2. Run the Docker with appropriate flags to run the test
-# 3. Upload the newly built Docker image
-#
-# in a way that is somewhat compatible with trampoline_v1.
-#
-# To run this script, first download few files from gcs to /dev/shm.
-# (/dev/shm is passed into the container as KOKORO_GFILE_DIR).
-#
-# gcloud storage cp gs://cloud-devrel-kokoro-resources/python-docs-samples/secrets_viewer_service_account.json /dev/shm
-# gcloud storage cp gs://cloud-devrel-kokoro-resources/python-docs-samples/automl_secrets.txt /dev/shm
-#
-# Then run the script.
-# .kokoro/trampoline_v2.sh
-#
-# These environment variables are required:
-# TRAMPOLINE_IMAGE: The docker image to use.
-# TRAMPOLINE_DOCKERFILE: The location of the Dockerfile.
-#
-# You can optionally change these environment variables:
-# TRAMPOLINE_IMAGE_UPLOAD:
-#     (true|false): Whether to upload the Docker image after the
-#                   successful builds.
-# TRAMPOLINE_BUILD_FILE: The script to run in the docker container.
-# TRAMPOLINE_WORKSPACE: The workspace path in the docker container.
-#                       Defaults to /workspace.
-# Potentially there are some repo specific envvars in .trampolinerc in
-# the project root.
-
-
-set -euo pipefail
-
-TRAMPOLINE_VERSION="2.0.5"
-
-if command -v tput >/dev/null && [[ -n "${TERM:-}" ]]; then
-  readonly IO_COLOR_RED="$(tput setaf 1)"
-  readonly IO_COLOR_GREEN="$(tput setaf 2)"
-  readonly IO_COLOR_YELLOW="$(tput setaf 3)"
-  readonly IO_COLOR_RESET="$(tput sgr0)"
-else
-  readonly IO_COLOR_RED=""
-  readonly IO_COLOR_GREEN=""
-  readonly IO_COLOR_YELLOW=""
-  readonly IO_COLOR_RESET=""
-fi
-
-function function_exists {
-    [ $(LC_ALL=C type -t $1)"" == "function" ]
-}
-
-# Logs a message using the given color. The first argument must be one
-# of the IO_COLOR_* variables defined above, such as
-# "${IO_COLOR_YELLOW}". The remaining arguments will be logged in the
-# given color. The log message will also have an RFC-3339 timestamp
-# prepended (in UTC). You can disable the color output by setting
-# TERM=vt100.
-function log_impl() {
-    local color="$1"
-    shift
-    local timestamp="$(date -u "+%Y-%m-%dT%H:%M:%SZ")"
-    echo "================================================================"
-    echo "${color}${timestamp}:" "$@" "${IO_COLOR_RESET}"
-    echo "================================================================"
-}
-
-# Logs the given message with normal coloring and a timestamp.
-function log() {
-  log_impl "${IO_COLOR_RESET}" "$@"
-}
-
-# Logs the given message in green with a timestamp.
-function log_green() {
-  log_impl "${IO_COLOR_GREEN}" "$@"
-}
-
-# Logs the given message in yellow with a timestamp.
-function log_yellow() {
-  log_impl "${IO_COLOR_YELLOW}" "$@"
-}
-
-# Logs the given message in red with a timestamp.
-function log_red() {
-  log_impl "${IO_COLOR_RED}" "$@"
-}
-
-readonly tmpdir=$(mktemp -d -t ci-XXXXXXXX)
-readonly tmphome="${tmpdir}/h"
-mkdir -p "${tmphome}"
-
-function cleanup() {
-    rm -rf "${tmpdir}"
-}
-trap cleanup EXIT
-
-RUNNING_IN_CI="${RUNNING_IN_CI:-false}"
-
-# The workspace in the container, defaults to /workspace.
-TRAMPOLINE_WORKSPACE="${TRAMPOLINE_WORKSPACE:-/workspace}"
-
-pass_down_envvars=(
-    # TRAMPOLINE_V2 variables.
-    # Tells scripts whether they are running as part of CI or not.
-    "RUNNING_IN_CI"
-    # Indicates which CI system we're in.
-    "TRAMPOLINE_CI"
-    # Indicates the version of the script.
-    "TRAMPOLINE_VERSION"
-)
-
-log_yellow "Building with Trampoline ${TRAMPOLINE_VERSION}"
-
-# Detect which CI systems we're in. If we're in any of the CI systems
-# we support, `RUNNING_IN_CI` will be true and `TRAMPOLINE_CI` will be
-# the name of the CI system. Both envvars will be passing down to the
-# container for telling which CI system we're in.
-if [[ -n "${KOKORO_BUILD_ID:-}" ]]; then
-    # descriptive env var for indicating it's on CI.
-    RUNNING_IN_CI="true"
-    TRAMPOLINE_CI="kokoro"
-    if [[ "${TRAMPOLINE_USE_LEGACY_SERVICE_ACCOUNT:-}" == "true" ]]; then
-	if [[ ! -f "${KOKORO_GFILE_DIR}/kokoro-trampoline.service-account.json" ]]; then
-	    log_red "${KOKORO_GFILE_DIR}/kokoro-trampoline.service-account.json does not exist. Did you forget to mount cloud-devrel-kokoro-resources/trampoline? Aborting."
-	    exit 1
-	fi
-	# This service account will be activated later.
-	TRAMPOLINE_SERVICE_ACCOUNT="${KOKORO_GFILE_DIR}/kokoro-trampoline.service-account.json"
-    else
-	if [[ "${TRAMPOLINE_VERBOSE:-}" == "true" ]]; then
-	    gcloud auth list
-	fi
-	log_yellow "Configuring Container Registry access"
-	gcloud auth configure-docker --quiet
-    fi
-    pass_down_envvars+=(
-	# KOKORO dynamic variables.
-	"KOKORO_BUILD_NUMBER"
-	"KOKORO_BUILD_ID"
-	"KOKORO_JOB_NAME"
-	"KOKORO_GIT_COMMIT"
-	"KOKORO_GITHUB_COMMIT"
-	"KOKORO_GITHUB_PULL_REQUEST_NUMBER"
-	"KOKORO_GITHUB_PULL_REQUEST_COMMIT"
-	# For FlakyBot
-	"KOKORO_GITHUB_COMMIT_URL"
-	"KOKORO_GITHUB_PULL_REQUEST_URL"
-    )
-elif [[ "${TRAVIS:-}" == "true" ]]; then
-    RUNNING_IN_CI="true"
-    TRAMPOLINE_CI="travis"
-    pass_down_envvars+=(
-	"TRAVIS_BRANCH"
-	"TRAVIS_BUILD_ID"
-	"TRAVIS_BUILD_NUMBER"
-	"TRAVIS_BUILD_WEB_URL"
-	"TRAVIS_COMMIT"
-	"TRAVIS_COMMIT_MESSAGE"
-	"TRAVIS_COMMIT_RANGE"
-	"TRAVIS_JOB_NAME"
-	"TRAVIS_JOB_NUMBER"
-	"TRAVIS_JOB_WEB_URL"
-	"TRAVIS_PULL_REQUEST"
-	"TRAVIS_PULL_REQUEST_BRANCH"
-	"TRAVIS_PULL_REQUEST_SHA"
-	"TRAVIS_PULL_REQUEST_SLUG"
-	"TRAVIS_REPO_SLUG"
-	"TRAVIS_SECURE_ENV_VARS"
-	"TRAVIS_TAG"
-    )
-elif [[ -n "${GITHUB_RUN_ID:-}" ]]; then
-    RUNNING_IN_CI="true"
-    TRAMPOLINE_CI="github-workflow"
-    pass_down_envvars+=(
-	"GITHUB_WORKFLOW"
-	"GITHUB_RUN_ID"
-	"GITHUB_RUN_NUMBER"
-	"GITHUB_ACTION"
-	"GITHUB_ACTIONS"
-	"GITHUB_ACTOR"
-	"GITHUB_REPOSITORY"
-	"GITHUB_EVENT_NAME"
-	"GITHUB_EVENT_PATH"
-	"GITHUB_SHA"
-	"GITHUB_REF"
-	"GITHUB_HEAD_REF"
-	"GITHUB_BASE_REF"
-    )
-elif [[ "${CIRCLECI:-}" == "true" ]]; then
-    RUNNING_IN_CI="true"
-    TRAMPOLINE_CI="circleci"
-    pass_down_envvars+=(
-	"CIRCLE_BRANCH"
-	"CIRCLE_BUILD_NUM"
-	"CIRCLE_BUILD_URL"
-	"CIRCLE_COMPARE_URL"
-	"CIRCLE_JOB"
-	"CIRCLE_NODE_INDEX"
-	"CIRCLE_NODE_TOTAL"
-	"CIRCLE_PREVIOUS_BUILD_NUM"
-	"CIRCLE_PROJECT_REPONAME"
-	"CIRCLE_PROJECT_USERNAME"
-	"CIRCLE_REPOSITORY_URL"
-	"CIRCLE_SHA1"
-	"CIRCLE_STAGE"
-	"CIRCLE_USERNAME"
-	"CIRCLE_WORKFLOW_ID"
-	"CIRCLE_WORKFLOW_JOB_ID"
-	"CIRCLE_WORKFLOW_UPSTREAM_JOB_IDS"
-	"CIRCLE_WORKFLOW_WORKSPACE_ID"
-    )
-fi
-
-# Configure the service account for pulling the docker image.
-function repo_root() {
-    local dir="$1"
-    while [[ ! -d "${dir}/.git" ]]; do
-	dir="$(dirname "$dir")"
-    done
-    echo "${dir}"
-}
-
-# Detect the project root. In CI builds, we assume the script is in
-# the git tree and traverse from there, otherwise, traverse from `pwd`
-# to find `.git` directory.
-if [[ "${RUNNING_IN_CI:-}" == "true" ]]; then
-    PROGRAM_PATH="$(realpath "$0")"
-    PROGRAM_DIR="$(dirname "${PROGRAM_PATH}")"
-    PROJECT_ROOT="$(repo_root "${PROGRAM_DIR}")"
-else
-    PROJECT_ROOT="$(repo_root $(pwd))"
-fi
-
-log_yellow "Changing to the project root: ${PROJECT_ROOT}."
-cd "${PROJECT_ROOT}"
-
-# To support relative path for `TRAMPOLINE_SERVICE_ACCOUNT`, we need
-# to use this environment variable in `PROJECT_ROOT`.
-if [[ -n "${TRAMPOLINE_SERVICE_ACCOUNT:-}" ]]; then
-
-    mkdir -p "${tmpdir}/gcloud"
-    gcloud_config_dir="${tmpdir}/gcloud"
-
-    log_yellow "Using isolated gcloud config: ${gcloud_config_dir}."
-    export CLOUDSDK_CONFIG="${gcloud_config_dir}"
-
-    log_yellow "Using ${TRAMPOLINE_SERVICE_ACCOUNT} for authentication."
-    gcloud auth activate-service-account \
-	   --key-file "${TRAMPOLINE_SERVICE_ACCOUNT}"
-    log_yellow "Configuring Container Registry access"
-    gcloud auth configure-docker --quiet
-fi
-
-required_envvars=(
-    # The basic trampoline configurations.
-    "TRAMPOLINE_IMAGE"
-    "TRAMPOLINE_BUILD_FILE"
-)
-
-if [[ -f "${PROJECT_ROOT}/.trampolinerc" ]]; then
-    source "${PROJECT_ROOT}/.trampolinerc"
-fi
-
-log_yellow "Checking environment variables."
-for e in "${required_envvars[@]}"
-do
-    if [[ -z "${!e:-}" ]]; then
-	log "Missing ${e} env var. Aborting."
-	exit 1
-    fi
-done
-
-# We want to support legacy style TRAMPOLINE_BUILD_FILE used with V1
-# script: e.g. "github/repo-name/.kokoro/run_tests.sh"
-TRAMPOLINE_BUILD_FILE="${TRAMPOLINE_BUILD_FILE#github/*/}"
-log_yellow "Using TRAMPOLINE_BUILD_FILE: ${TRAMPOLINE_BUILD_FILE}"
-
-# ignore error on docker operations and test execution
-set +e
-
-log_yellow "Preparing Docker image."
-# We only download the docker image in CI builds.
-if [[ "${RUNNING_IN_CI:-}" == "true" ]]; then
-    # Download the docker image specified by `TRAMPOLINE_IMAGE`
-
-    # We may want to add --max-concurrent-downloads flag.
-
-    log_yellow "Start pulling the Docker image: ${TRAMPOLINE_IMAGE}."
-    if docker pull "${TRAMPOLINE_IMAGE}"; then
-	log_green "Finished pulling the Docker image: ${TRAMPOLINE_IMAGE}."
-	has_image="true"
-    else
-	log_red "Failed pulling the Docker image: ${TRAMPOLINE_IMAGE}."
-	has_image="false"
-    fi
-else
-    # For local run, check if we have the image.
-    if docker images "${TRAMPOLINE_IMAGE}:latest" | grep "${TRAMPOLINE_IMAGE}"; then
-	has_image="true"
-    else
-	has_image="false"
-    fi
-fi
-
-
-# The default user for a Docker container has uid 0 (root). To avoid
-# creating root-owned files in the build directory we tell docker to
-# use the current user ID.
-user_uid="$(id -u)"
-user_gid="$(id -g)"
-user_name="$(id -un)"
-
-# To allow docker in docker, we add the user to the docker group in
-# the host os.
-docker_gid=$(cut -d: -f3 < <(getent group docker))
-
-update_cache="false"
-if [[ "${TRAMPOLINE_DOCKERFILE:-none}" != "none" ]]; then
-    # Build the Docker image from the source.
-    context_dir=$(dirname "${TRAMPOLINE_DOCKERFILE}")
-    docker_build_flags=(
-	"-f" "${TRAMPOLINE_DOCKERFILE}"
-	"-t" "${TRAMPOLINE_IMAGE}"
-	"--build-arg" "UID=${user_uid}"
-	"--build-arg" "USERNAME=${user_name}"
-    )
-    if [[ "${has_image}" == "true" ]]; then
-	docker_build_flags+=("--cache-from" "${TRAMPOLINE_IMAGE}")
-    fi
-
-    log_yellow "Start building the docker image."
-    if [[ "${TRAMPOLINE_VERBOSE:-false}" == "true" ]]; then
-	echo "docker build" "${docker_build_flags[@]}" "${context_dir}"
-    fi
-
-    # ON CI systems, we want to suppress docker build logs, only
-    # output the logs when it fails.
-    if [[ "${RUNNING_IN_CI:-}" == "true" ]]; then
-	if docker build "${docker_build_flags[@]}" "${context_dir}" \
-		  > "${tmpdir}/docker_build.log" 2>&1; then
-	    if [[ "${TRAMPOLINE_VERBOSE:-}" == "true" ]]; then
-		cat "${tmpdir}/docker_build.log"
-	    fi
-
-	    log_green "Finished building the docker image."
-	    update_cache="true"
-	else
-	    log_red "Failed to build the Docker image, aborting."
-	    log_yellow "Dumping the build logs:"
-	    cat "${tmpdir}/docker_build.log"
-	    exit 1
-	fi
-    else
-	if docker build "${docker_build_flags[@]}" "${context_dir}"; then
-	    log_green "Finished building the docker image."
-	    update_cache="true"
-	else
-	    log_red "Failed to build the Docker image, aborting."
-	    exit 1
-	fi
-    fi
-else
-    if [[ "${has_image}" != "true" ]]; then
-	log_red "We do not have ${TRAMPOLINE_IMAGE} locally, aborting."
-	exit 1
-    fi
-fi
-
-# We use an array for the flags so they are easier to document.
-docker_flags=(
-    # Remove the container after it exists.
-    "--rm"
-
-    # Use the host network.
-    "--network=host"
-
-    # Run in privileged mode. We are not using docker for sandboxing or
-    # isolation, just for packaging our dev tools.
-    "--privileged"
-
-    # Run the docker script with the user id. Because the docker image gets to
-    # write in ${PWD} you typically want this to be your user id.
-    # To allow docker in docker, we need to use docker gid on the host.
-    "--user" "${user_uid}:${docker_gid}"
-
-    # Pass down the USER.
-    "--env" "USER=${user_name}"
-
-    # Mount the project directory inside the Docker container.
-    "--volume" "${PROJECT_ROOT}:${TRAMPOLINE_WORKSPACE}"
-    "--workdir" "${TRAMPOLINE_WORKSPACE}"
-    "--env" "PROJECT_ROOT=${TRAMPOLINE_WORKSPACE}"
-
-    # Mount the temporary home directory.
-    "--volume" "${tmphome}:/h"
-    "--env" "HOME=/h"
-
-    # Allow docker in docker.
-    "--volume" "/var/run/docker.sock:/var/run/docker.sock"
-
-    # Mount the /tmp so that docker in docker can mount the files
-    # there correctly.
-    "--volume" "/tmp:/tmp"
-    # Pass down the KOKORO_GFILE_DIR and KOKORO_KEYSTORE_DIR
-    # TODO(tmatsuo): This part is not portable.
-    "--env" "TRAMPOLINE_SECRET_DIR=/secrets"
-    "--volume" "${KOKORO_GFILE_DIR:-/dev/shm}:/secrets/gfile"
-    "--env" "KOKORO_GFILE_DIR=/secrets/gfile"
-    "--volume" "${KOKORO_KEYSTORE_DIR:-/dev/shm}:/secrets/keystore"
-    "--env" "KOKORO_KEYSTORE_DIR=/secrets/keystore"
-)
-
-# Add an option for nicer output if the build gets a tty.
-if [[ -t 0 ]]; then
-    docker_flags+=("-it")
-fi
-
-# Passing down env vars
-for e in "${pass_down_envvars[@]}"
-do
-    if [[ -n "${!e:-}" ]]; then
-	docker_flags+=("--env" "${e}=${!e}")
-    fi
-done
-
-# If arguments are given, all arguments will become the commands run
-# in the container, otherwise run TRAMPOLINE_BUILD_FILE.
-if [[ $# -ge 1 ]]; then
-    log_yellow "Running the given commands '" "${@:1}" "' in the container."
-    readonly commands=("${@:1}")
-    if [[ "${TRAMPOLINE_VERBOSE:-}" == "true" ]]; then
-	echo docker run "${docker_flags[@]}" "${TRAMPOLINE_IMAGE}" "${commands[@]}"
-    fi
-    docker run "${docker_flags[@]}" "${TRAMPOLINE_IMAGE}" "${commands[@]}"
-else
-    log_yellow "Running the tests in a Docker container."
-    docker_flags+=("--entrypoint=${TRAMPOLINE_BUILD_FILE}")
-    if [[ "${TRAMPOLINE_VERBOSE:-}" == "true" ]]; then
-	echo docker run "${docker_flags[@]}" "${TRAMPOLINE_IMAGE}"
-    fi
-    docker run "${docker_flags[@]}" "${TRAMPOLINE_IMAGE}"
-fi
-
-
-test_retval=$?
-
-if [[ ${test_retval} -eq 0 ]]; then
-    log_green "Build finished with ${test_retval}"
-else
-    log_red "Build finished with ${test_retval}"
-fi
-
-# Only upload it when the test passes.
-if [[ "${update_cache}" == "true" ]] && \
-       [[ $test_retval == 0 ]] && \
-       [[ "${TRAMPOLINE_IMAGE_UPLOAD:-false}" == "true" ]]; then
-    log_yellow "Uploading the Docker image."
-    if docker push "${TRAMPOLINE_IMAGE}"; then
-	log_green "Finished uploading the Docker image."
-    else
-	log_red "Failed uploading the Docker image."
-    fi
-    # Call trampoline_after_upload_hook if it's defined.
-    if function_exists trampoline_after_upload_hook; then
-	trampoline_after_upload_hook
-    fi
-
-fi
-
-exit "${test_retval}"
diff --git a/.trampolinerc b/.trampolinerc
deleted file mode 100644
index 00801523..00000000
--- a/.trampolinerc
+++ /dev/null
@@ -1,61 +0,0 @@
-# Copyright 2024 Google LLC
-#
-# Licensed 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.
-
-# Add required env vars here.
-required_envvars+=(
-)
-
-# Add env vars which are passed down into the container here.
-pass_down_envvars+=(
-    "NOX_SESSION"
-    ###############
-    # Docs builds
-    ###############
-    "STAGING_BUCKET"
-    "V2_STAGING_BUCKET"
-    ##################
-    # Samples builds
-    ##################
-    "INSTALL_LIBRARY_FROM_SOURCE"
-    "RUN_TESTS_SESSION"
-    "BUILD_SPECIFIC_GCLOUD_PROJECT"
-    # Target directories.
-    "RUN_TESTS_DIRS"
-    # The nox session to run.
-    "RUN_TESTS_SESSION"
-)
-
-# Prevent unintentional override on the default image.
-if [[ "${TRAMPOLINE_IMAGE_UPLOAD:-false}" == "true" ]] && \
-   [[ -z "${TRAMPOLINE_IMAGE:-}" ]]; then
-   echo "Please set TRAMPOLINE_IMAGE if you want to upload the Docker image."
-   exit 1
-fi
-
-# Define the default value if it makes sense.
-if [[ -z "${TRAMPOLINE_IMAGE_UPLOAD:-}" ]]; then
-    TRAMPOLINE_IMAGE_UPLOAD=""
-fi
-
-if [[ -z "${TRAMPOLINE_IMAGE:-}" ]]; then
-    TRAMPOLINE_IMAGE=""
-fi
-
-if [[ -z "${TRAMPOLINE_DOCKERFILE:-}" ]]; then
-    TRAMPOLINE_DOCKERFILE=""
-fi
-
-if [[ -z "${TRAMPOLINE_BUILD_FILE:-}" ]]; then
-    TRAMPOLINE_BUILD_FILE=""
-fi
diff --git a/DEVELOPING.md b/DEVELOPING.md
index 23b5a027..6e42182b 100644
--- a/DEVELOPING.md
+++ b/DEVELOPING.md
@@ -1,16 +1,21 @@
 # Developing
 
-To get started, create a fresh `venv` or conda environment. Then, update `pip`
-and install the required dev dependencies within the new environment.
+To get started, create a fresh virtual environment (`uv venv` or `python -m venv`)
+and install the development and test dependencies:
 
 ```sh
+# Using uv (recommended):
+uv pip install -r requirements-dev.txt
+
+# Or using pip:
 pip install -U pip
 pip install -r requirements-dev.txt
-
-# Install the test dependencies to run unit and integration tests.
-pip install -r requirements-test.txt
 ```
 
+`requirements-dev.txt` installs the package in editable mode with the `[client]`
+and `[interactive]` extras (`-e .[client,interactive]`), along with
+`requirements-test.txt` (`pytest`, `pytest-xdist`), `ipywidgets`, and `pyink`.
+
 # Linting/formatting
 
 We use `pyink` to lint/format the code. To apply changes to your local
@@ -24,21 +29,30 @@ This will ensure that your changes pass CI code linting.
 
 # Testing
 
-We use `pytest` for testing. To run _all_ tests, simply run `pytest`. For this
-to work, you have to have your ambient environment configured to correctly
-resolve all configuration details such as GCP project, region, etc.
+We use `pytest` for testing. To run unit tests in parallel:
 
-To make testing more deterministic, it is instead recommended to specify
-configuration details on the command line. For example:
+```sh
+pytest -n auto tests/unit
+```
+
+To run integration tests against Google Cloud Managed Spark, provide your
+project and region (plus optional subnet or service account configuration):
 
 ```sh
 env \
   GOOGLE_CLOUD_PROJECT='project-id' \
   GOOGLE_CLOUD_REGION='us-central1' \
-  MANAGED_SPARK_CONNECT_SUBNET='subnet-id' \
-  pytest --tb=auto -v
+  pytest tests/integration --tb=short -v
 ```
 
+`MANAGED_SPARK_CONNECT_SUBNET` is optional (defaults to `"default"`) and accepts
+either a short subnetwork name in `GOOGLE_CLOUD_REGION` (e.g., `default`) or a
+full `projects/.../regions/.../subnetworks/...` URI. Integration tests share a
+session-scoped default cluster and automatically set safety TTLs
+(`MANAGED_SPARK_CONNECT_TTL_SECONDS=1800` and
+`MANAGED_SPARK_CONNECT_IDLE_TTL_SECONDS=600`), so run `tests/integration`
+without `-n auto` (~8 minutes total).
+
 ## Testing the Interactive Extras
 
 The dependencies of the interactive extras (`explore_dataframe`, `%dpip`,
@@ -50,44 +64,32 @@ and `traitlets` are importable and will fail, not skip, if they are not.
 Missing dependencies are simulated within the tests themselves, by setting
 `sys.modules[name] = None`.
 
-`tests/local` runs `%%sparksql` end to end against local classic and Spark
-Connect sessions. It needs no Google Cloud project, only Java 17 or 21 (set
-`JAVA_HOME` if your default `java` is another version, or the tests skip), and
-starts a JVM in-process, so run it on its own rather than under `xdist`:
+`tests/local` runs `%%sparksql` and local session creation end to end against
+local classic and Spark Connect sessions. It needs no Google Cloud project, only
+Java 17 or 21 (set `JAVA_HOME` if your default `java` is another version, or the
+tests skip), and starts a JVM in-process, so run it on its own rather than under
+`xdist`:
 
 ```sh
 pytest tests/local
 ```
 
-The integration tests in particular can take a while to run. To speed up the
-testing cycle, you can run them in parallel. You can do so using the `xdist`
-plugin by setting the `-n` flag to the number of parallel runners you want to
-use. This will be set automatically if you set it to `auto`. For example:
-
-```sh
-env \
-  GOOGLE_CLOUD_PROJECT='project-id' \
-  GOOGLE_CLOUD_REGION='us-central1' \
-  MANAGED_SPARK_CONNECT_SUBNET='subnet-id' \
-  MANAGED_SPARK_CONNECT_SERVICE_ACCOUNT='service@account.test' \
-  pytest -n auto --tb=auto -v
-```
-
 ## Testing Against Full PySpark
 
 `requirements-dev.txt` installs the lightweight `pyspark-client`, under which
-`tests/local` and the integration tests that need a local JVM skip. To run them,
-switch the environment to full `pyspark`. Both packages write to the same
-`pyspark` directory, so always uninstall both before installing either:
+`tests/local` skips. To run local JVM Spark tests, switch the environment (or
+create a separate virtual environment) with the `[spark]` extra. Both packages
+write to the same `pyspark` directory, so always uninstall both before
+installing either:
 
 ```sh
-pip uninstall -y pyspark-client pyspark
-pip install 'pyspark[connect]~=4.0.0'
+uv pip uninstall pyspark-client pyspark
+uv pip install -e '.[spark,interactive]' -r requirements-test.txt
 ```
 
 To switch back to `pyspark-client`:
 
 ```sh
-pip uninstall -y pyspark-client pyspark
-pip install 'pyspark-client~=4.0.0'
+uv pip uninstall pyspark-client pyspark
+uv pip install -e '.[client,interactive]' -r requirements-test.txt
 ```
diff --git a/cloudbuild/Dockerfile b/cloudbuild/Dockerfile
deleted file mode 100644
index d95d53b2..00000000
--- a/cloudbuild/Dockerfile
+++ /dev/null
@@ -1,14 +0,0 @@
-FROM python:3.10-bookworm
-
-WORKDIR /opt/tests
-# NOTE: We copy the requirements files explicitly _before_ installing
-# dependencies to allow layer caching. Afterward, we copy the rest of the build
-# sources. The intention is to reuse the preinstalled dependencies when
-# individual source files change.
-COPY requirements-dev.txt ./
-COPY requirements-test.txt ./
-RUN python3 -m pip install -U pip
-RUN python3 -m pip install --no-cache-dir -r requirements-dev.txt -r requirements-test.txt
-COPY . .
-RUN python setup.py sdist bdist_wheel egg_info
-RUN pip install .
diff --git a/cloudbuild/cloudbuild.yaml b/cloudbuild/cloudbuild.yaml
deleted file mode 100644
index bb56bfb1..00000000
--- a/cloudbuild/cloudbuild.yaml
+++ /dev/null
@@ -1,13 +0,0 @@
-steps:
-  # Build a container image which bakes in a source code snapshot and built
-  # distribution artifacts.
-  - name: 'gcr.io/cloud-builders/docker'
-    id: 'build-container-image'
-    args: ['build', '--tag=gcr.io/${PROJECT_ID}/managed-spark-connect/managed-spark-connect-presubmit:${BUILD_ID}', -f, 'cloudbuild/Dockerfile', '.']
-  # Run all unit tests
-  - name: 'gcr.io/${PROJECT_ID}/managed-spark-connect/managed-spark-connect-presubmit:${BUILD_ID}'
-    id: 'run-unit-tests'
-    waitFor: ['build-container-image']
-    entrypoint: 'pytest'
-    args: ['-n', 'auto', 'tests/unit']
-timeout: 600s
diff --git a/pyproject.toml b/pyproject.toml
index d9506e28..56261e7d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,3 +1,7 @@
+[build-system]
+requires = ["setuptools>=72.0"]
+build-backend = "setuptools.build_meta"
+
 [tool.pyink]
 line-length = 80  # (default is 88)
 pyink-indentation = 4 #  (default is 4)
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 772be7ce..2cf8d595 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -1,13 +1,4 @@
-google-api-core>=2.19
-google-cloud-dataproc>=5.18
-google-colabsqlviz>=0.4.0
-ipython~=9.1
+-e .[client,interactive]
+-r requirements-test.txt
 ipywidgets>=8.0.0
-packaging>=20.0
 pyink~=24.0
-pyspark-client~=4.0.0
-setuptools>=72.0
-sqlparse>=0.5.0
-tqdm>=4.67
-traitlets>=5.1
-websockets>=14.0
diff --git a/setup.cfg b/setup.cfg
index e339f3e0..e078a91d 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,6 +1,3 @@
-[bdist_wheel]
-universal = 1
-
 [check-manifest]
 ignore =
     .github/**
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 6b25a5ee..197b6a48 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -11,17 +11,26 @@
 # 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.
+import logging
 import os
+import signal
+
 import certifi
 from google.api_core import client_options
+from google.api_core import exceptions as core_exceptions
 from google.cloud.dataproc_v1 import (
     SessionControllerClient,
     SessionTemplateControllerClient,
+    TerminateSessionRequest,
 )
 from google.cloud.managed_spark_connect import ManagedSparkSession
 import pytest
 
+logger = logging.getLogger(__name__)
+
 _SERVICE_ACCOUNT_KEY_FILE_ = "service_account_key.json"
+_DEFAULT_FALLBACK_REGIONS = ("us-central1", "us-east1", "us-west1")
+_terminated_session_ids = set()
 
 
 def is_ci_environment():
@@ -29,14 +38,122 @@ def is_ci_environment():
     return os.getenv("CI") == "true" or os.getenv("GITHUB_ACTIONS") == "true"
 
 
+def _qualify_subnetwork_uri(test_project, test_region, test_subnet):
+    """Qualify a short subnetwork name into a regional Compute subnetwork URI."""
+    if "/" in test_subnet:
+        return test_subnet
+    return (
+        f"projects/{test_project}/regions/{test_region}/subnetworks/"
+        f"{test_subnet}"
+    )
+
+
+def _candidate_regions(
+    primary_region,
+    test_subnet="default",
+    fallback_regions_env=None,
+    custom_endpoint=None,
+):
+    """Return ordered list of regions to try when creating the shared session.
+
+    When test_subnet is a full resource URI (containing '/') or a custom regional
+    API endpoint is configured, only primary_region is returned because the URI
+    or endpoint is already bound to a single region.
+    """
+    primary = primary_region or "us-central1"
+    if "/" in test_subnet or custom_endpoint:
+        return [primary]
+
+    if fallback_regions_env is None:
+        fallback_regions_env = os.getenv("GOOGLE_CLOUD_FALLBACK_REGIONS")
+    if fallback_regions_env is not None:
+        fallbacks = [
+            r.strip() for r in fallback_regions_env.split(",") if r.strip()
+        ]
+    else:
+        fallbacks = list(_DEFAULT_FALLBACK_REGIONS)
+
+    candidates = [primary]
+    for region in fallbacks:
+        if region not in candidates:
+            candidates.append(region)
+    return candidates
+
+
+def _apply_region_to_env(test_project, region, test_subnet):
+    """Update environment variables for the active integration test region."""
+    os.environ["GOOGLE_CLOUD_REGION"] = region
+    os.environ["MANAGED_SPARK_CONNECT_SUBNET"] = _qualify_subnetwork_uri(
+        test_project, region, test_subnet
+    )
+
+
+def _is_service_unavailable(exc):
+    """Return True if exc or its chained cause/context is 503 ServiceUnavailable."""
+    current = exc
+    visited = set()
+    while current is not None and id(current) not in visited:
+        visited.add(id(current))
+        if isinstance(current, core_exceptions.ServiceUnavailable):
+            return True
+        current = current.__cause__ or current.__context__
+    return False
+
+
+def _terminate_active_session():
+    """Best-effort, idempotent termination of any active or in-flight ManagedSparkSession."""
+    session_id = ManagedSparkSession._active_s8s_session_id
+    if not session_id or session_id in _terminated_session_ids:
+        _clear_session_state()
+        return
+    _terminated_session_ids.add(session_id)
+    try:
+        try:
+            active = (
+                ManagedSparkSession.getActiveSession()
+                or ManagedSparkSession._default_session
+            )
+            if active is not None:
+                active.stop(terminate=True)
+                return
+        except Exception:
+            pass
+
+        project_id = ManagedSparkSession._project_id or os.getenv(
+            "GOOGLE_CLOUD_PROJECT"
+        )
+        region = ManagedSparkSession._region or os.getenv("GOOGLE_CLOUD_REGION")
+        if project_id and region:
+            try:
+                endpoint = os.getenv(
+                    "GOOGLE_CLOUD_DATAPROC_API_ENDPOINT",
+                    f"{region}-dataproc.googleapis.com",
+                )
+                opts = client_options.ClientOptions(api_endpoint=endpoint)
+                client = SessionControllerClient(client_options=opts)
+                name = (
+                    f"projects/{project_id}/locations/{region}/sessions/"
+                    f"{session_id}"
+                )
+                client.terminate_session(TerminateSessionRequest(name=name))
+            except Exception:
+                pass
+    finally:
+        _clear_session_state()
+
+
 @pytest.fixture(scope="session")
 def test_project():
     return os.getenv("GOOGLE_CLOUD_PROJECT")
 
 
 @pytest.fixture(scope="session")
-def test_region():
-    return os.getenv("GOOGLE_CLOUD_REGION")
+def _region_state():
+    return {
+        "region": os.getenv("GOOGLE_CLOUD_REGION") or "us-central1",
+        "session_clients": {},
+        "template_clients": {},
+    }
 
 
 @pytest.fixture(scope="session")
@@ -62,17 +179,7 @@ def test_subnet():
 
 
 @pytest.fixture(scope="session")
-def test_subnetwork_uri(test_project, test_region, test_subnet):
-    if "/" in test_subnet:
-        return test_subnet
-    return (
-        f"projects/{test_project}/regions/{test_region}/subnetworks/"
-        f"{test_subnet}"
-    )
-
-
-@pytest.fixture(scope="session")
-def os_environment(auth_type, test_project, test_region, test_subnetwork_uri):
+def os_environment(auth_type, test_project, _region_state, test_subnet):
     original_environment = dict(os.environ)
     if os.path.isfile(_SERVICE_ACCOUNT_KEY_FILE_):
         os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = (
@@ -81,7 +188,7 @@ def os_environment(auth_type, test_project, test_region, test_subnetwork_uri):
     os.environ["MANAGED_SPARK_CONNECT_AUTH_TYPE"] = auth_type
     if auth_type == "END_USER_CREDENTIALS":
         os.environ.pop("MANAGED_SPARK_CONNECT_SERVICE_ACCOUNT", None)
-    os.environ["MANAGED_SPARK_CONNECT_SUBNET"] = test_subnetwork_uri
+    _apply_region_to_env(test_project, _region_state["region"], test_subnet)
     # Set short safety TTLs so any session leaked by a crashed or cancelled
     # test run auto-terminates after 10 minutes of inactivity (or 30 minutes
     # total) instead of the 24-hour server default.
@@ -90,14 +197,38 @@ def os_environment(auth_type, test_project, test_region, test_subnetwork_uri):
     # Add SSL certificate fix
     os.environ["SSL_CERT_FILE"] = certifi.where()
     os.environ["REQUESTS_CA_BUNDLE"] = certifi.where()
-    yield os.environ
-    for key in list(os.environ.keys()):
-        if key not in original_environment:
-            del os.environ[key]
-    os.environ.update(original_environment)
 
+    previous_sigterm = signal.getsignal(signal.SIGTERM)
 
-@pytest.fixture(scope="session")
+    def _handle_sigterm(signum, frame):
+        _terminate_active_session()
+        if callable(previous_sigterm):
+            previous_sigterm(signum, frame)
+        raise KeyboardInterrupt("Received SIGTERM during integration tests")
+
+    signal.signal(signal.SIGTERM, _handle_sigterm)
+    try:
+        yield os.environ
+    finally:
+        _terminate_active_session()
+        signal.signal(signal.SIGTERM, previous_sigterm)
+        for key in list(os.environ.keys()):
+            if key not in original_environment:
+                del os.environ[key]
+        os.environ.update(original_environment)
+
+
+@pytest.fixture
+def test_region(_region_state):
+    return _region_state["region"]
+
+
+@pytest.fixture
+def test_subnetwork_uri(test_project, test_region, test_subnet):
+    return _qualify_subnetwork_uri(test_project, test_region, test_subnet)
+
+
+@pytest.fixture
 def api_endpoint(test_region):
     return os.getenv(
         "GOOGLE_CLOUD_DATAPROC_API_ENDPOINT",
@@ -105,19 +236,31 @@ def api_endpoint(test_region):
     )
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture
 def test_client_options(api_endpoint, os_environment):
     return client_options.ClientOptions(api_endpoint=api_endpoint)
 
 
-@pytest.fixture(scope="session")
-def session_controller_client(test_client_options):
-    return SessionControllerClient(client_options=test_client_options)
+@pytest.fixture
+def session_controller_client(_region_state, test_region, test_client_options):
+    clients = _region_state["session_clients"]
+    if test_region not in clients:
+        clients[test_region] = SessionControllerClient(
+            client_options=test_client_options
+        )
+    return clients[test_region]
 
 
-@pytest.fixture(scope="session")
-def session_template_controller_client(test_client_options):
-    return SessionTemplateControllerClient(client_options=test_client_options)
+@pytest.fixture
+def session_template_controller_client(
+    _region_state, test_region, test_client_options
+):
+    clients = _region_state["template_clients"]
+    if test_region not in clients:
+        clients[test_region] = SessionTemplateControllerClient(
+            client_options=test_client_options
+        )
+    return clients[test_region]
 
 
 def _capture_session_state():
@@ -177,27 +320,66 @@ def _clear_session_state():
         pass
 
 
+def _create_session_with_region_fallback(
+    test_project, _region_state, test_subnet
+):
+    """Create the shared ManagedSparkSession, falling back across regions on 503 stockouts."""
+    candidates = _candidate_regions(
+        _region_state["region"],
+        test_subnet=test_subnet,
+        custom_endpoint=os.getenv("GOOGLE_CLOUD_DATAPROC_API_ENDPOINT"),
+    )
+    last_error = None
+    for index, candidate_region in enumerate(candidates):
+        _clear_session_state()
+        _region_state["region"] = candidate_region
+        _apply_region_to_env(test_project, candidate_region, test_subnet)
+        try:
+            return (
+                ManagedSparkSession.builder.projectId(test_project)
+                .location(candidate_region)
+                .getOrCreate()
+            )
+        except BaseException as exc:
+            _terminate_active_session()
+            if not isinstance(exc, Exception) or not _is_service_unavailable(
+                exc
+            ):
+                raise
+            last_error = exc
+            if index + 1 < len(candidates):
+                next_region = candidates[index + 1]
+                logger.warning(
+                    "Managed Spark session creation in %s failed with 503"
+                    " ServiceUnavailable (%s); falling back to %s",
+                    candidate_region,
+                    exc,
+                    next_region,
+                )
+            else:
+                raise
+    raise last_error
+
+
 @pytest.fixture(scope="session")
-def _shared_connect_session(test_project, test_region, os_environment):
+def _shared_connect_session(
+    test_project, _region_state, test_subnet, os_environment
+):
     """Session-scoped default ManagedSparkSession shared by non-destructive tests.
 
     Created with default builder settings (multi-node) so out-of-the-box user
     defaults, Python UDF execution on remote executors, and %dpip package
     distribution are exercised against a standard cluster.
     """
-    _clear_session_state()
-    session = (
-        ManagedSparkSession.builder.projectId(test_project)
-        .location(test_region)
-        .getOrCreate()
+    session = _create_session_with_region_fallback(
+        test_project, _region_state, test_subnet
     )
     shared_state = _capture_session_state()
-    yield shared_state
-    _restore_session_state(shared_state)
     try:
-        session.stop(terminate=True)
-    except Exception:
-        pass
+        yield shared_state
+    finally:
+        _restore_session_state(shared_state)
+        _terminate_active_session()
 
 
 @pytest.fixture
@@ -209,7 +391,7 @@ def connect_session(_shared_connect_session):
 
 
 @pytest.fixture
-def session_name(test_project, test_region, connect_session):
+def session_name(test_project, connect_session, test_region):
     return (
         f"projects/{test_project}/locations/{test_region}/sessions/"
         f"{ManagedSparkSession._active_s8s_session_id}"
@@ -226,13 +408,5 @@ def isolated_session_state(os_environment):
     finally:
         leaked_id = ManagedSparkSession._active_s8s_session_id
         if leaked_id and leaked_id != saved_state["session_id"]:
-            try:
-                active = (
-                    ManagedSparkSession.getActiveSession()
-                    or ManagedSparkSession._default_session
-                )
-                if active is not None:
-                    active.stop(terminate=True)
-            except Exception:
-                pass
+            _terminate_active_session()
         _restore_session_state(saved_state)
diff --git a/tests/unit/test_integration_conftest.py b/tests/unit/test_integration_conftest.py
new file mode 100644
index 00000000..24ea26dc
--- /dev/null
+++ b/tests/unit/test_integration_conftest.py
@@ -0,0 +1,165 @@
+# Copyright 2026 Google LLC
+#
+# Licensed 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.
+import os
+import unittest
+from unittest import mock
+
+from google.api_core import exceptions as core_exceptions
+from google.cloud.managed_spark_connect import ManagedSparkSession
+from tests.integration import conftest as integration_conftest
+
+
+class IntegrationConftestTest(unittest.TestCase):
+
+    def setUp(self):
+        super().setUp()
+        self._saved_env = dict(os.environ)
+        os.environ.pop("GOOGLE_CLOUD_FALLBACK_REGIONS", None)
+        integration_conftest._terminated_session_ids.clear()
+        integration_conftest._clear_session_state()
+
+    def tearDown(self):
+        integration_conftest._terminated_session_ids.clear()
+        integration_conftest._clear_session_state()
+        for key in list(os.environ.keys()):
+            if key not in self._saved_env:
+                del os.environ[key]
+        os.environ.update(self._saved_env)
+        super().tearDown()
+
+    def test_candidate_regions_with_short_subnet(self):
+        candidates = integration_conftest._candidate_regions(
+            "us-east1", test_subnet="custom-subnet", fallback_regions_env=""
+        )
+        self.assertEqual(candidates, ["us-east1"])
+
+        default_candidates = integration_conftest._candidate_regions(
+            "us-east1", test_subnet="custom-subnet"
+        )
+        self.assertEqual(
+            default_candidates, ["us-east1", "us-central1", "us-west1"]
+        )
+
+    def test_candidate_regions_disables_fallback_for_full_subnet_uri(self):
+        full_uri = "projects/my-proj/regions/us-east1/subnetworks/custom-subnet"
+        candidates = integration_conftest._candidate_regions(
+            "us-east1", test_subnet=full_uri
+        )
+        self.assertEqual(candidates, ["us-east1"])
+
+    def test_candidate_regions_disables_fallback_for_custom_endpoint(self):
+        candidates = integration_conftest._candidate_regions(
+            "us-east1",
+            test_subnet="default",
+            custom_endpoint="us-east1-custom-dataproc.googleapis.com",
+        )
+        self.assertEqual(candidates, ["us-east1"])
+
+    @mock.patch.object(ManagedSparkSession.Builder, "getOrCreate")
+    def test_create_session_with_region_fallback_switches_on_wrapped_503(
+        self, mock_get_or_create
+    ):
+        fake_session = mock.MagicMock()
+        wrapped_503 = RuntimeError("Error while creating Managed Spark Session")
+        wrapped_503.__cause__ = core_exceptions.ServiceUnavailable(
+            "Failed to create main node pool"
+        )
+        mock_get_or_create.side_effect = [wrapped_503, fake_session]
+        region_state = {
+            "region": "us-central1",
+            "session_clients": {},
+            "template_clients": {},
+        }
+        os.environ.pop("GOOGLE_CLOUD_DATAPROC_API_ENDPOINT", None)
+        os.environ["GOOGLE_CLOUD_FALLBACK_REGIONS"] = "us-central1,us-east1"
+
+        result = integration_conftest._create_session_with_region_fallback(
+            "test-proj", region_state, "custom-subnet"
+        )
+
+        self.assertIs(result, fake_session)
+        self.assertEqual(mock_get_or_create.call_count, 2)
+        self.assertEqual(region_state["region"], "us-east1")
+        self.assertEqual(os.environ["GOOGLE_CLOUD_REGION"], "us-east1")
+        self.assertEqual(
+            os.environ["MANAGED_SPARK_CONNECT_SUBNET"],
+            "projects/test-proj/regions/us-east1/subnetworks/custom-subnet",
+        )
+
+    @mock.patch.object(ManagedSparkSession.Builder, "getOrCreate")
+    def test_create_session_with_region_fallback_does_not_retry_non_503_runtime_error(
+        self, mock_get_or_create
+    ):
+        non_503 = RuntimeError("Error while creating Managed Spark Session")
+        non_503.__cause__ = ValueError("Invalid configuration")
+        mock_get_or_create.side_effect = non_503
+        region_state = {
+            "region": "us-central1",
+            "session_clients": {},
+            "template_clients": {},
+        }
+        os.environ["GOOGLE_CLOUD_FALLBACK_REGIONS"] = "us-central1,us-east1"
+
+        with self.assertRaises(RuntimeError):
+            integration_conftest._create_session_with_region_fallback(
+                "test-proj", region_state, "default"
+            )
+        self.assertEqual(mock_get_or_create.call_count, 1)
+
+    @mock.patch.object(ManagedSparkSession.Builder, "getOrCreate")
+    def test_create_session_with_region_fallback_raises_when_exhausted(
+        self, mock_get_or_create
+    ):
+        wrapped_503 = RuntimeError("Error while creating Managed Spark Session")
+        wrapped_503.__cause__ = core_exceptions.ServiceUnavailable(
+            "Stockout everywhere"
+        )
+        mock_get_or_create.side_effect = wrapped_503
+        region_state = {
+            "region": "us-central1",
+            "session_clients": {},
+            "template_clients": {},
+        }
+        os.environ["GOOGLE_CLOUD_FALLBACK_REGIONS"] = "us-central1,us-east1"
+
+        with self.assertRaises(RuntimeError):
+            integration_conftest._create_session_with_region_fallback(
+                "test-proj", region_state, "default"
+            )
+        self.assertEqual(mock_get_or_create.call_count, 2)
+
+    @mock.patch.object(integration_conftest, "SessionControllerClient")
+    def test_terminate_active_session_in_flight_via_controller_is_idempotent(
+        self, mock_client_cls
+    ):
+        ManagedSparkSession._active_s8s_session_id = "sc-in-flight-123"
+        ManagedSparkSession._project_id = "test-proj"
+        ManagedSparkSession._region = "us-east1"
+
+        integration_conftest._terminate_active_session()
+        # Second call with same session ID should be a no-op
+        ManagedSparkSession._active_s8s_session_id = "sc-in-flight-123"
+        integration_conftest._terminate_active_session()
+
+        mock_client_cls.return_value.terminate_session.assert_called_once()
+        req = mock_client_cls.return_value.terminate_session.call_args[0][0]
+        self.assertEqual(
+            req.name,
+            "projects/test-proj/locations/us-east1/sessions/sc-in-flight-123",
+        )
+        self.assertIsNone(ManagedSparkSession._active_s8s_session_id)
+
+
+if __name__ == "__main__":
+    unittest.main()