diff --git a/.github/workflows/build_cluster_http_path.py b/.github/workflows/build_cluster_http_path.py
index 264018019..76ad456ab 100644
--- a/.github/workflows/build_cluster_http_path.py
+++ b/.github/workflows/build_cluster_http_path.py
@@ -1,16 +1,24 @@
 import os
 import re
 
-workspace_re = re.compile(r"^.*-(\d+)\..*$")
-hostname = os.getenv("DBT_DATABRICKS_HOST_NAME", "")
-matches = workspace_re.match(hostname)
-if matches:
-    workspace_id = matches.group(1)
-    print(workspace_id)
+spog_native = os.getenv("TEST_PECO_SPOG_NATIVE") == "1"
+spog_workspace_id = os.getenv("TEST_PECO_SPOG_WORKSPACE_ID")
+
+if spog_native:
+    if not spog_workspace_id:
+        raise RuntimeError("TEST_PECO_SPOG_NATIVE requires TEST_PECO_SPOG_WORKSPACE_ID.")
+    workspace_id = spog_workspace_id
+else:
+    workspace_re = re.compile(r"^.*-(\d+)\..*$")
+    hostname = os.getenv("DBT_DATABRICKS_HOST_NAME", "")
+    matches = workspace_re.match(hostname)
+    workspace_id = matches.group(1) if matches else ""
+
 cluster_id = os.getenv("TEST_PECO_CLUSTER_ID")
 uc_cluster_id = os.getenv("TEST_PECO_UC_CLUSTER_ID")
-http_path = f"sql/protocolv1/o/{workspace_id}/{cluster_id}"
-uc_http_path = f"sql/protocolv1/o/{workspace_id}/{uc_cluster_id}"
+suffix = f"?o={workspace_id}" if spog_native else ""
+http_path = f"sql/protocolv1/o/{workspace_id}/{cluster_id}{suffix}"
+uc_http_path = f"sql/protocolv1/o/{workspace_id}/{uc_cluster_id}{suffix}"
 
 # https://stackoverflow.com/a/72225291/5093960
 env_file = os.getenv("GITHUB_ENV", "")
diff --git a/.github/workflows/integration-min-deps.yml b/.github/workflows/integration-min-deps.yml
index be0ce8ea3..0776fbbff 100644
--- a/.github/workflows/integration-min-deps.yml
+++ b/.github/workflows/integration-min-deps.yml
@@ -290,11 +290,6 @@ jobs:
         run: |
           mkdir -p logs
           DBT_TEST_USER=notnecessaryformosttests@example.com \
-          DBT_DATABRICKS_LOCATION_ROOT="$DBT_DATABRICKS_LOCATION_ROOT" \
-          DBT_DATABRICKS_HOST_NAME="$DBT_DATABRICKS_HOST_NAME" \
-          DBT_DATABRICKS_UC_CLUSTER_HTTP_PATH="$DBT_DATABRICKS_UC_CLUSTER_HTTP_PATH" \
-          DBT_DATABRICKS_CLIENT_ID="$DBT_DATABRICKS_CLIENT_ID" \
-          DBT_DATABRICKS_CLIENT_SECRET="$DBT_DATABRICKS_CLIENT_SECRET" \
           xargs -r hatch -v run min-deps:pytest \
             --color=yes -v \
             --profile databricks_uc_cluster \
@@ -389,11 +384,6 @@ jobs:
         run: |
           mkdir -p logs
           DBT_TEST_USER=notnecessaryformosttests@example.com \
-          DBT_DATABRICKS_LOCATION_ROOT="$DBT_DATABRICKS_LOCATION_ROOT" \
-          DBT_DATABRICKS_HOST_NAME="$DBT_DATABRICKS_HOST_NAME" \
-          DBT_DATABRICKS_UC_CLUSTER_HTTP_PATH="$DBT_DATABRICKS_UC_CLUSTER_HTTP_PATH" \
-          DBT_DATABRICKS_CLIENT_ID="$DBT_DATABRICKS_CLIENT_ID" \
-          DBT_DATABRICKS_CLIENT_SECRET="$DBT_DATABRICKS_CLIENT_SECRET" \
           xargs -r hatch -v run min-deps:pytest \
             --color=yes -v \
             --profile databricks_uc_sql_endpoint \
@@ -486,11 +476,7 @@ jobs:
         run: |
           mkdir -p logs
           DBT_TEST_USER=notnecessaryformosttests@example.com \
-          DBT_DATABRICKS_LOCATION_ROOT="$DBT_DATABRICKS_LOCATION_ROOT" \
-          DBT_DATABRICKS_HOST_NAME="$DBT_DATABRICKS_HOST_NAME" \
           DBT_DATABRICKS_HTTP_PATH="$DBT_DATABRICKS_CLUSTER_HTTP_PATH" \
-          DBT_DATABRICKS_CLIENT_ID="$DBT_DATABRICKS_CLIENT_ID" \
-          DBT_DATABRICKS_CLIENT_SECRET="$DBT_DATABRICKS_CLIENT_SECRET" \
           xargs -r hatch -v run min-deps:pytest \
             --color=yes -v \
             --profile databricks_cluster \
diff --git a/.github/workflows/integration-spog.yml b/.github/workflows/integration-spog.yml
new file mode 100644
index 000000000..4a1dea9ad
--- /dev/null
+++ b/.github/workflows/integration-spog.yml
@@ -0,0 +1,325 @@
+# SPOG Integration Tests for dbt-databricks.
+#
+# Mirrors integration.yml's matrix (prepare-shards + 3 sharded functional
+# jobs) on a weekly Sunday schedule, but with the host name and http_path
+# wired to the SPOG vanity host + ?o= so every test path exercises SPOG
+# routing. PR-targeting / status-reporting logic is omitted — this is a
+# scheduled smoke, not a PR gate.
+name: SPOG Integration Tests
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: "30 21 * * 0"  # Weekly: Sunday 21:30 UTC (Monday 03:00 IST).
+
+permissions:
+  id-token: write
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  prepare-shards:
+    runs-on:
+      group: databricks-protected-runner-group
+      labels: linux-ubuntu-latest
+    env:
+      UV_FROZEN: "1"
+    steps:
+      - name: Check out the repository
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5  # v4
+
+      - name: Setup JFrog PyPI Proxy
+        uses: ./.github/actions/setup-jfrog-pypi
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065  # v5
+        with:
+          python-version: "3.10"
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@38f3f104447c67c051c4a08e39b64a148898af3a  # v4
+        with:
+          cache-local-path: ~/.cache/uv
+
+      - name: Install Hatch
+        uses: pypa/hatch@257e27e51a6a5616ed08a39a408a21c35c9931bc  # install
+
+      - name: Collect tests and assign to shards
+        run: |
+          set -euo pipefail
+          mkdir -p shard-assignments
+          declare -A NUM_SHARDS=(
+            [databricks_cluster]=2
+            [databricks_uc_cluster]=3
+            [databricks_uc_sql_endpoint]=3
+          )
+          for PROFILE in "${!NUM_SHARDS[@]}"; do
+            (
+              hatch run pytest --collect-only -q --profile "$PROFILE" tests/functional 2>&1 \
+                | grep "::" \
+                > "shard-assignments/${PROFILE}-collected.txt"
+            ) &
+          done
+          wait
+          for PROFILE in "${!NUM_SHARDS[@]}"; do
+            python3 scripts/shard_assign.py \
+              --profile "$PROFILE" \
+              --num-shards "${NUM_SHARDS[$PROFILE]}" \
+              --input "shard-assignments/${PROFILE}-collected.txt" \
+              --output-dir shard-assignments \
+              --algo lpt_historical_time \
+              --timings .github/test_timings.json
+          done
+
+      - name: Upload shard assignments
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02  # v4
+        with:
+          name: shard-assignments-spog
+          path: shard-assignments/
+          retention-days: 5
+
+  uc-cluster:
+    needs: prepare-shards
+    strategy:
+      fail-fast: false
+      matrix:
+        shard: [0, 1, 2]
+    runs-on:
+      group: databricks-protected-runner-group
+      labels: linux-ubuntu-latest
+    environment: azure-prod
+    env:
+      DBT_DATABRICKS_HOST_NAME: ${{ secrets.TEST_PECO_SPOG_HOST }}
+      DBT_DATABRICKS_CLIENT_ID: ${{ secrets.TEST_PECO_SP_ID }}
+      DBT_DATABRICKS_CLIENT_SECRET: ${{ secrets.TEST_PECO_SP_SECRET }}
+      DBT_DATABRICKS_UC_INITIAL_CATALOG: peco
+      DBT_DATABRICKS_LOCATION_ROOT: ${{ secrets.TEST_PECO_EXTERNAL_LOCATION }}test
+      TEST_PECO_SPOG_WORKSPACE_ID: ${{ secrets.TEST_PECO_SPOG_WORKSPACE_ID }}
+      TEST_PECO_UC_CLUSTER_ID: ${{ secrets.TEST_PECO_UC_CLUSTER_ID }}
+      TEST_PECO_SPOG_NATIVE: "1"
+      UV_FROZEN: "1"
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5  # v4
+
+      - name: Setup JFrog PyPI Proxy
+        uses: ./.github/actions/setup-jfrog-pypi
+
+      - name: Set up python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065  # v5
+        with:
+          python-version: "3.10"
+
+      - name: Get http path from environment
+        run: python .github/workflows/build_cluster_http_path.py
+        shell: sh
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@38f3f104447c67c051c4a08e39b64a148898af3a  # v4
+        with:
+          cache-local-path: ~/.cache/uv
+
+      - name: Install Hatch
+        uses: pypa/hatch@257e27e51a6a5616ed08a39a408a21c35c9931bc  # install
+
+      - name: Download shard assignments
+        uses: actions/download-artifact@65a9edc5881444af0b9093a5e628f2fe47ea3b2e  # v4.1.7
+        with:
+          name: shard-assignments-spog
+          path: shard-assignments/
+
+      - name: Resolve test list for this shard
+        run: |
+          set -euo pipefail
+          SHARD_FILE="shard-assignments/databricks_uc_cluster-shard-${{ matrix.shard }}.txt"
+          if [ ! -s "$SHARD_FILE" ]; then
+            echo "::error::Shard file missing or empty: $SHARD_FILE"
+            exit 1
+          fi
+          echo "SHARD_TESTS_FILE=$SHARD_FILE" >> "$GITHUB_ENV"
+          echo "Files in shard ${{ matrix.shard }}: $(wc -l < "$SHARD_FILE")"
+
+      - name: Run UC Cluster Functional Tests (shard ${{ matrix.shard }})
+        run: |
+          mkdir -p logs
+          DBT_TEST_USER=notnecessaryformosttests@example.com \
+          xargs -r hatch -v run pytest \
+            --color=yes -v \
+            --profile databricks_uc_cluster \
+            -n 10 --dist=loadfile \
+            --reruns 1 --reruns-delay 120 \
+            --junitxml=logs/junit-uc-cluster-shard-${{ matrix.shard }}.xml \
+            < "$SHARD_TESTS_FILE"
+
+      - name: Upload UC Cluster Test Logs
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02  # v4
+        with:
+          name: spog-uc-cluster-logs-shard-${{ matrix.shard }}
+          path: logs/
+          retention-days: 14
+
+  uc-sql-endpoint:
+    needs: prepare-shards
+    strategy:
+      fail-fast: false
+      matrix:
+        shard: [0, 1, 2]
+    runs-on:
+      group: databricks-protected-runner-group
+      labels: linux-ubuntu-latest
+    environment: azure-prod
+    env:
+      DBT_DATABRICKS_HOST_NAME: ${{ secrets.TEST_PECO_SPOG_HOST }}
+      DBT_DATABRICKS_CLIENT_ID: ${{ secrets.TEST_PECO_SP_ID }}
+      DBT_DATABRICKS_CLIENT_SECRET: ${{ secrets.TEST_PECO_SP_SECRET }}
+      DBT_DATABRICKS_HTTP_PATH: ${{ secrets.TEST_PECO_WAREHOUSE_HTTP_PATH }}?o=${{ secrets.TEST_PECO_SPOG_WORKSPACE_ID }}
+      DBT_DATABRICKS_UC_INITIAL_CATALOG: peco
+      DBT_DATABRICKS_LOCATION_ROOT: ${{ secrets.TEST_PECO_EXTERNAL_LOCATION }}test
+      TEST_PECO_SPOG_WORKSPACE_ID: ${{ secrets.TEST_PECO_SPOG_WORKSPACE_ID }}
+      TEST_PECO_UC_CLUSTER_ID: ${{ secrets.TEST_PECO_UC_CLUSTER_ID }}
+      TEST_PECO_SPOG_NATIVE: "1"
+      UV_FROZEN: "1"
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5  # v4
+
+      - name: Setup JFrog PyPI Proxy
+        uses: ./.github/actions/setup-jfrog-pypi
+
+      - name: Set up python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065  # v5
+        with:
+          python-version: "3.10"
+
+      - name: Get http path from environment
+        run: python .github/workflows/build_cluster_http_path.py
+        shell: sh
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@38f3f104447c67c051c4a08e39b64a148898af3a  # v4
+        with:
+          cache-local-path: ~/.cache/uv
+
+      - name: Install Hatch
+        uses: pypa/hatch@257e27e51a6a5616ed08a39a408a21c35c9931bc  # install
+
+      - name: Download shard assignments
+        uses: actions/download-artifact@65a9edc5881444af0b9093a5e628f2fe47ea3b2e  # v4.1.7
+        with:
+          name: shard-assignments-spog
+          path: shard-assignments/
+
+      - name: Resolve test list for this shard
+        run: |
+          set -euo pipefail
+          SHARD_FILE="shard-assignments/databricks_uc_sql_endpoint-shard-${{ matrix.shard }}.txt"
+          if [ ! -s "$SHARD_FILE" ]; then
+            echo "::error::Shard file missing or empty: $SHARD_FILE"
+            exit 1
+          fi
+          echo "SHARD_TESTS_FILE=$SHARD_FILE" >> "$GITHUB_ENV"
+          echo "Files in shard ${{ matrix.shard }}: $(wc -l < "$SHARD_FILE")"
+
+      - name: Run Sql Endpoint Functional Tests (shard ${{ matrix.shard }})
+        run: |
+          mkdir -p logs
+          DBT_TEST_USER=notnecessaryformosttests@example.com \
+          xargs -r hatch -v run pytest \
+            --color=yes -v \
+            --profile databricks_uc_sql_endpoint \
+            -n 10 --dist=loadfile \
+            --reruns 1 --reruns-delay 120 \
+            --junitxml=logs/junit-uc-sql-endpoint-shard-${{ matrix.shard }}.xml \
+            < "$SHARD_TESTS_FILE"
+
+      - name: Upload SQL Endpoint Test Logs
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02  # v4
+        with:
+          name: spog-uc-sql-endpoint-logs-shard-${{ matrix.shard }}
+          path: logs/
+          retention-days: 14
+
+  cluster:
+    needs: prepare-shards
+    strategy:
+      fail-fast: false
+      matrix:
+        shard: [0, 1]
+    runs-on:
+      group: databricks-protected-runner-group
+      labels: linux-ubuntu-latest
+    environment: azure-prod
+    env:
+      DBT_DATABRICKS_HOST_NAME: ${{ secrets.TEST_PECO_SPOG_HOST }}
+      DBT_DATABRICKS_CLIENT_ID: ${{ secrets.TEST_PECO_SP_ID }}
+      DBT_DATABRICKS_CLIENT_SECRET: ${{ secrets.TEST_PECO_SP_SECRET }}
+      DBT_DATABRICKS_LOCATION_ROOT: ${{ secrets.TEST_PECO_EXTERNAL_LOCATION }}test
+      TEST_PECO_SPOG_WORKSPACE_ID: ${{ secrets.TEST_PECO_SPOG_WORKSPACE_ID }}
+      TEST_PECO_CLUSTER_ID: ${{ secrets.TEST_PECO_CLUSTER_ID }}
+      TEST_PECO_SPOG_NATIVE: "1"
+      UV_FROZEN: "1"
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5  # v4
+
+      - name: Setup JFrog PyPI Proxy
+        uses: ./.github/actions/setup-jfrog-pypi
+
+      - name: Set up python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065  # v5
+        with:
+          python-version: "3.10"
+
+      - name: Get http path from environment
+        run: python .github/workflows/build_cluster_http_path.py
+        shell: sh
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@38f3f104447c67c051c4a08e39b64a148898af3a  # v4
+        with:
+          cache-local-path: ~/.cache/uv
+
+      - name: Install Hatch
+        uses: pypa/hatch@257e27e51a6a5616ed08a39a408a21c35c9931bc  # install
+
+      - name: Download shard assignments
+        uses: actions/download-artifact@65a9edc5881444af0b9093a5e628f2fe47ea3b2e  # v4.1.7
+        with:
+          name: shard-assignments-spog
+          path: shard-assignments/
+
+      - name: Resolve test list for this shard
+        run: |
+          set -euo pipefail
+          SHARD_FILE="shard-assignments/databricks_cluster-shard-${{ matrix.shard }}.txt"
+          if [ ! -s "$SHARD_FILE" ]; then
+            echo "::error::Shard file missing or empty: $SHARD_FILE"
+            exit 1
+          fi
+          echo "SHARD_TESTS_FILE=$SHARD_FILE" >> "$GITHUB_ENV"
+          echo "Files in shard ${{ matrix.shard }}: $(wc -l < "$SHARD_FILE")"
+
+      - name: Run Cluster Functional Tests (shard ${{ matrix.shard }})
+        run: |
+          mkdir -p logs
+          DBT_TEST_USER=notnecessaryformosttests@example.com \
+          DBT_DATABRICKS_HTTP_PATH="$DBT_DATABRICKS_CLUSTER_HTTP_PATH" \
+          xargs -r hatch -v run pytest \
+            --color=yes -v \
+            --profile databricks_cluster \
+            -n 10 --dist=loadfile \
+            --reruns 1 --reruns-delay 120 \
+            --junitxml=logs/junit-cluster-shard-${{ matrix.shard }}.xml \
+            < "$SHARD_TESTS_FILE"
+
+      - name: Upload Cluster Test Logs
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02  # v4
+        with:
+          name: spog-cluster-logs-shard-${{ matrix.shard }}
+          path: logs/
+          retention-days: 14
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index 63447f462..16501b5eb 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -248,6 +248,8 @@ jobs:
       DBT_DATABRICKS_UC_INITIAL_CATALOG: peco
       DBT_DATABRICKS_LOCATION_ROOT: ${{ secrets.TEST_PECO_EXTERNAL_LOCATION }}test
       TEST_PECO_UC_CLUSTER_ID: ${{ secrets.TEST_PECO_UC_CLUSTER_ID }}
+      TEST_PECO_SPOG_HOST: ${{ secrets.TEST_PECO_SPOG_HOST }}
+      TEST_PECO_SPOG_WORKSPACE_ID: ${{ secrets.TEST_PECO_SPOG_WORKSPACE_ID }}
       UV_FROZEN: "1"
     steps:
       - name: Check out repository
@@ -303,11 +305,6 @@ jobs:
         run: |
           mkdir -p logs
           DBT_TEST_USER=notnecessaryformosttests@example.com \
-          DBT_DATABRICKS_LOCATION_ROOT="$DBT_DATABRICKS_LOCATION_ROOT" \
-          DBT_DATABRICKS_HOST_NAME="$DBT_DATABRICKS_HOST_NAME" \
-          DBT_DATABRICKS_UC_CLUSTER_HTTP_PATH="$DBT_DATABRICKS_UC_CLUSTER_HTTP_PATH" \
-          DBT_DATABRICKS_CLIENT_ID="$DBT_DATABRICKS_CLIENT_ID" \
-          DBT_DATABRICKS_CLIENT_SECRET="$DBT_DATABRICKS_CLIENT_SECRET" \
           xargs -r hatch -v run pytest \
             --color=yes -v \
             --profile databricks_uc_cluster \
@@ -349,6 +346,8 @@ jobs:
       DBT_DATABRICKS_UC_INITIAL_CATALOG: peco
       DBT_DATABRICKS_LOCATION_ROOT: ${{ secrets.TEST_PECO_EXTERNAL_LOCATION }}test
       TEST_PECO_UC_CLUSTER_ID: ${{ secrets.TEST_PECO_UC_CLUSTER_ID }}
+      TEST_PECO_SPOG_HOST: ${{ secrets.TEST_PECO_SPOG_HOST }}
+      TEST_PECO_SPOG_WORKSPACE_ID: ${{ secrets.TEST_PECO_SPOG_WORKSPACE_ID }}
       UV_FROZEN: "1"
     steps:
       - name: Check out repository
@@ -404,11 +403,6 @@ jobs:
         run: |
           mkdir -p logs
           DBT_TEST_USER=notnecessaryformosttests@example.com \
-          DBT_DATABRICKS_LOCATION_ROOT="$DBT_DATABRICKS_LOCATION_ROOT" \
-          DBT_DATABRICKS_HOST_NAME="$DBT_DATABRICKS_HOST_NAME" \
-          DBT_DATABRICKS_UC_CLUSTER_HTTP_PATH="$DBT_DATABRICKS_UC_CLUSTER_HTTP_PATH" \
-          DBT_DATABRICKS_CLIENT_ID="$DBT_DATABRICKS_CLIENT_ID" \
-          DBT_DATABRICKS_CLIENT_SECRET="$DBT_DATABRICKS_CLIENT_SECRET" \
           xargs -r hatch -v run pytest \
             --color=yes -v \
             --profile databricks_uc_sql_endpoint \
@@ -448,6 +442,8 @@ jobs:
       DBT_DATABRICKS_CLIENT_SECRET: ${{ secrets.TEST_PECO_SP_SECRET }}
       TEST_PECO_CLUSTER_ID: ${{ secrets.TEST_PECO_CLUSTER_ID }}
       DBT_DATABRICKS_LOCATION_ROOT: ${{ secrets.TEST_PECO_EXTERNAL_LOCATION }}test
+      TEST_PECO_SPOG_HOST: ${{ secrets.TEST_PECO_SPOG_HOST }}
+      TEST_PECO_SPOG_WORKSPACE_ID: ${{ secrets.TEST_PECO_SPOG_WORKSPACE_ID }}
       UV_FROZEN: "1"
     steps:
       - name: Check out repository
@@ -503,11 +499,7 @@ jobs:
         run: |
           mkdir -p logs
           DBT_TEST_USER=notnecessaryformosttests@example.com \
-          DBT_DATABRICKS_LOCATION_ROOT="$DBT_DATABRICKS_LOCATION_ROOT" \
-          DBT_DATABRICKS_HOST_NAME="$DBT_DATABRICKS_HOST_NAME" \
           DBT_DATABRICKS_HTTP_PATH="$DBT_DATABRICKS_CLUSTER_HTTP_PATH" \
-          DBT_DATABRICKS_CLIENT_ID="$DBT_DATABRICKS_CLIENT_ID" \
-          DBT_DATABRICKS_CLIENT_SECRET="$DBT_DATABRICKS_CLIENT_SECRET" \
           xargs -r hatch -v run pytest \
             --color=yes -v \
             --profile databricks_cluster \
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 125eb9806..83b6cafad 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,12 +3,14 @@
 ### Features
 
 - Expose `job_id`, `job_run_id`, and `task_run_id` from the Databricks Jobs `dbt_task` runtime in `adapter_response`, enabling correlation between dbt runs and Databricks workflow executions via `run_results.json` ([#1451](https://github.com/databricks/dbt-databricks/pull/1451) closes [#722](https://github.com/databricks/dbt-databricks/issues/722))
+- Add support for SPOG (Single Point of Gateway) hosts: account-level vanity URLs with `?o=<workspace-id>` in `http_path` route correctly for both data-plane (SQL) and control-plane (REST/Jobs/Workspace API) traffic. Requires `databricks-sql-connector >= 4.2.6` and `databricks-sdk >= 0.76.0`. ([#1479](https://github.com/databricks/dbt-databricks/pull/1479))
 
 ### Fixes
 
 - Fix missing f-string prefix in `JobRunsApi.submit` debug log ([#1471](https://github.com/databricks/dbt-databricks/pull/1471))
 - Fix capability-branching macros falling through to their legacy path at parse/compile time on SQL warehouses. The parse-time stub of `has_dbr_capability` now returns `True` on warehouse profiles for capabilities flagged `sql_warehouse_supported`, so macros select the modern branch during compilation instead of the legacy fallback. ([#1449](https://github.com/databricks/dbt-databricks/pull/1449) closes [#1331](https://github.com/databricks/dbt-databricks/issues/1331))
 - Fix snapshots not applying `databricks_tags` on columns ([#1442](https://github.com/databricks/dbt-databricks/pull/1442) closes [#1441](https://github.com/databricks/dbt-databricks/issues/1441))
+- `EXTRACT_CLUSTER_ID_FROM_HTTP_PATH_REGEX` now stops the capture at `?` / `&`, so any trailing query string on `http_path` no longer corrupts the extracted cluster id. Latent issue on legacy hosts; the fix unblocks SPOG cluster paths.
 
 ### Under the Hood
 
diff --git a/dbt/adapters/databricks/connections.py b/dbt/adapters/databricks/connections.py
index e442ab5a5..b6825edd6 100644
--- a/dbt/adapters/databricks/connections.py
+++ b/dbt/adapters/databricks/connections.py
@@ -54,6 +54,7 @@
 )
 from dbt.adapters.databricks.logging import logger
 from dbt.adapters.databricks.python_models.run_tracking import PythonRunTracker
+from dbt.adapters.databricks.spog.decision import check_spog_preconditions
 from dbt.adapters.databricks.utils import QueryTagsUtils, is_cluster_http_path, redact_credentials
 
 if TYPE_CHECKING:
@@ -483,6 +484,17 @@ def open(cls, connection: Connection) -> Connection:
 
         cls.credentials_manager = creds.authenticate()
 
+        # SPOG decision matrix: collect every http_path in play (default +
+        # per-compute) and validate them against the host's discovery probe.
+        # Raises DbtConfigError on misconfiguration; no-op on legacy hosts.
+        compute_paths = [
+            p for cfg in (creds.compute or {}).values() if cfg and (p := cfg.get("http_path"))
+        ]
+        check_spog_preconditions(
+            host=creds.host or "",
+            http_paths=[databricks_connection.http_path, *compute_paths],
+        )
+
         # Get merged query tags if we have query header context
         query_header_context = getattr(databricks_connection, "_query_header_context", None)
         merged_query_tags = {}
diff --git a/dbt/adapters/databricks/credentials.py b/dbt/adapters/databricks/credentials.py
index 6e57ac8c3..e1561cc08 100644
--- a/dbt/adapters/databricks/credentials.py
+++ b/dbt/adapters/databricks/credentials.py
@@ -15,10 +15,12 @@
 from databricks.sdk.core import Config, CredentialsProvider
 from dbt.adapters.databricks.global_state import GlobalState
 from dbt.adapters.databricks.logging import logger
+from dbt.adapters.databricks.spog.capabilities import sdk_supports_workspace_id
+from dbt.adapters.databricks.spog.extract import extract_workspace_id
 
 CATALOG_KEY_IN_SESSION_PROPERTIES = "databricks.catalog"
 DBT_DATABRICKS_INVOCATION_ENV_REGEX = re.compile("^[A-z0-9\\-]+$")
-EXTRACT_CLUSTER_ID_FROM_HTTP_PATH_REGEX = re.compile(r"/?sql/protocolv1/o/\d+/(.*)")
+EXTRACT_CLUSTER_ID_FROM_HTTP_PATH_REGEX = re.compile(r"/?sql/protocolv1/o/\d+/([^?&]+)")
 DBT_DATABRICKS_HTTP_SESSION_HEADERS = "DBT_DATABRICKS_HTTP_SESSION_HEADERS"
 
 REDIRECT_URL = "http://localhost:8020"
@@ -277,6 +279,7 @@ class DatabricksCredentialManager(DataClassDictMixin):
     oauth_scopes: list[str] = field(default_factory=lambda: SCOPES)
     token: Optional[str] = None
     auth_type: Optional[str] = None
+    workspace_id: Optional[str] = None
 
     @classmethod
     def create_from(cls, credentials: DatabricksCredentials) -> "DatabricksCredentialManager":
@@ -290,44 +293,62 @@ def create_from(cls, credentials: DatabricksCredentials) -> "DatabricksCredentia
             oauth_redirect_url=credentials.oauth_redirect_url or REDIRECT_URL,
             oauth_scopes=credentials.oauth_scopes or SCOPES,
             auth_type=credentials.auth_type,
+            workspace_id=extract_workspace_id(credentials.http_path),
         )
 
+    def _config_kwargs(self, **base: Any) -> dict[str, Any]:
+        """Conditionally add workspace_id to Config kwargs when the SDK supports it.
+
+        SPOG (account-level vanity hosts) needs `workspace_id` plumbed into the
+        SDK Config so REST calls carry the X-Databricks-Org-Id header. We only
+        add it when (a) one was extracted from http_path's `?o=` and (b) the
+        installed SDK exposes the field.
+        """
+        if self.workspace_id and sdk_supports_workspace_id():
+            base["workspace_id"] = self.workspace_id
+        return base
+
     def authenticate_with_pat(self) -> Config:
-        return Config(
-            host=self.host,
-            token=self.token,
-        )
+        return Config(**self._config_kwargs(host=self.host, token=self.token))
 
     def authenticate_with_oauth_m2m(self) -> Config:
         return Config(
-            host=self.host,
-            client_id=self.client_id,
-            client_secret=self.client_secret,
-            auth_type="oauth-m2m",
+            **self._config_kwargs(
+                host=self.host,
+                client_id=self.client_id,
+                client_secret=self.client_secret,
+                auth_type="oauth-m2m",
+            )
         )
 
     def authenticate_with_external_browser(self) -> Config:
         return Config(
-            host=self.host,
-            client_id=self.client_id,
-            client_secret=self.client_secret,
-            auth_type="external-browser",
+            **self._config_kwargs(
+                host=self.host,
+                client_id=self.client_id,
+                client_secret=self.client_secret,
+                auth_type="external-browser",
+            )
         )
 
     def legacy_authenticate_with_azure_client_secret(self) -> Config:
         return Config(
-            host=self.host,
-            azure_client_id=self.client_id,
-            azure_client_secret=self.client_secret,
-            auth_type="azure-client-secret",
+            **self._config_kwargs(
+                host=self.host,
+                azure_client_id=self.client_id,
+                azure_client_secret=self.client_secret,
+                auth_type="azure-client-secret",
+            )
         )
 
     def authenticate_with_azure_client_secret(self) -> Config:
         return Config(
-            host=self.host,
-            azure_client_id=self.azure_client_id,
-            azure_client_secret=self.azure_client_secret,
-            auth_type="azure-client-secret",
+            **self._config_kwargs(
+                host=self.host,
+                azure_client_id=self.azure_client_id,
+                azure_client_secret=self.azure_client_secret,
+                auth_type="azure-client-secret",
+            )
         )
 
     def __post_init__(self) -> None:
diff --git a/dbt/adapters/databricks/impl.py b/dbt/adapters/databricks/impl.py
index cc9be7a5b..dd94e46cd 100644
--- a/dbt/adapters/databricks/impl.py
+++ b/dbt/adapters/databricks/impl.py
@@ -51,9 +51,11 @@
     DatabricksConnectionManager,
     DatabricksDBTConnection,
 )
+from dbt.adapters.databricks.credentials import DatabricksCredentials
 from dbt.adapters.databricks.dbr_capabilities import DBRCapabilities, DBRCapability
 from dbt.adapters.databricks.global_state import GlobalState
 from dbt.adapters.databricks.handle import SqlUtils
+from dbt.adapters.databricks.logging import logger
 from dbt.adapters.databricks.python_models.python_submissions import (
     AllPurposeClusterPythonJobHelper,
     JobClusterPythonJobHelper,
@@ -87,6 +89,12 @@
 )
 from dbt.adapters.databricks.relation_configs.tblproperties import TblPropertiesConfig
 from dbt.adapters.databricks.relation_configs.view import ViewConfig
+from dbt.adapters.databricks.spog import probe
+from dbt.adapters.databricks.spog.capabilities import (
+    connector_supports_spog,
+    sdk_supports_workspace_id,
+)
+from dbt.adapters.databricks.spog.extract import extract_workspace_id
 from dbt.adapters.databricks.utils import (
     get_first_row,
     handle_missing_objects,
@@ -1107,6 +1115,52 @@ def resolve_file_format(self, config: BaseConfig) -> str:
                 return constants.DELTA_FILE_FORMAT
         return config.get("file_format", default=constants.DELTA_FILE_FORMAT)
 
+    def debug_query(self) -> None:
+        """Override for DebugTask: emit SPOG diagnostic block, then `select 1`."""
+        creds = cast(DatabricksCredentials, self.config.credentials)
+        DatabricksAdapter.debug_emit_spog_block(
+            host=creds.host or "",
+            http_path=creds.http_path or "",
+        )
+        self.execute("select 1 as id")
+
+    @staticmethod
+    def debug_emit_spog_block(host: str, http_path: str) -> None:
+        """Print a SPOG status block: host_type, workspace_id, dep versions.
+
+        Visible in `dbt debug` output. Makes "is SPOG working here?" a
+        one-command answer for support escalations.
+        """
+        if not host:
+            return
+
+        def _pkg_version_or_missing(name: str) -> str:
+            try:
+                return metadata.version(name)
+            except metadata.PackageNotFoundError:
+                return "<not installed>"
+
+        host_metadata = probe.probe_host(host)
+        is_spog = host_metadata.host_type == "unified"
+        workspace_id = extract_workspace_id(http_path)
+        connector_v = _pkg_version_or_missing("databricks-sql-connector")
+        sdk_v = _pkg_version_or_missing("databricks-sdk")
+        connector_ok = connector_supports_spog()
+        sdk_ok = sdk_supports_workspace_id()
+
+        logger.info(
+            f"  SPOG host (host_type={host_metadata.host_type!r}): {'yes' if is_spog else 'no'}"
+        )
+        logger.info(f"  workspace_id (from ?o= in http_path): {workspace_id!r}")
+        logger.info(
+            f"  databricks-sql-connector version: {connector_v} "
+            f"({'supported' if connector_ok else 'NOT supported (>= 4.2.6 required for SPOG)'})"
+        )
+        logger.info(
+            f"  databricks-sdk version: {sdk_v} "
+            f"({'supported' if sdk_ok else 'NOT supported (>= 0.76.0 required for SPOG)'})"
+        )
+
 
 @dataclass(frozen=True)
 class RelationAPIBase(ABC, Generic[DatabricksRelationConfig]):
diff --git a/dbt/adapters/databricks/spog/__init__.py b/dbt/adapters/databricks/spog/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/dbt/adapters/databricks/spog/capabilities.py b/dbt/adapters/databricks/spog/capabilities.py
new file mode 100644
index 000000000..e28756a14
--- /dev/null
+++ b/dbt/adapters/databricks/spog/capabilities.py
@@ -0,0 +1,26 @@
+"""Detect whether the installed connector and SDK support SPOG."""
+
+from functools import cache
+from importlib.metadata import PackageNotFoundError
+from importlib.metadata import version as _pkg_version
+
+from packaging.version import Version
+
+from databricks.sdk.core import Config
+
+CONNECTOR_SPOG_MIN = Version("4.2.6")
+
+
+@cache
+def connector_supports_spog() -> bool:
+    """True iff the installed databricks-sql-connector is >= 4.2.6."""
+    try:
+        return Version(_pkg_version("databricks-sql-connector")) >= CONNECTOR_SPOG_MIN
+    except PackageNotFoundError:
+        return False
+
+
+@cache
+def sdk_supports_workspace_id() -> bool:
+    """True iff the installed databricks-sdk exposes a `workspace_id` attribute on Config."""
+    return hasattr(Config, "workspace_id")
diff --git a/dbt/adapters/databricks/spog/decision.py b/dbt/adapters/databricks/spog/decision.py
new file mode 100644
index 000000000..0051a5bce
--- /dev/null
+++ b/dbt/adapters/databricks/spog/decision.py
@@ -0,0 +1,67 @@
+"""Decision matrix for SPOG preconditions, called from connection.open()."""
+
+from collections.abc import Iterable
+from typing import Optional
+
+from dbt_common.exceptions import DbtConfigError
+
+from dbt.adapters.databricks.logging import logger
+from dbt.adapters.databricks.spog import probe
+from dbt.adapters.databricks.spog.capabilities import (
+    connector_supports_spog,
+    sdk_supports_workspace_id,
+)
+from dbt.adapters.databricks.spog.extract import extract_workspace_id
+
+
+def check_spog_preconditions(
+    *,
+    host: str,
+    http_paths: Iterable[str],
+) -> Optional[str]:
+    """Validate SPOG-related configuration.
+
+    Returns the extracted workspace_id (the value of ``?o=`` in http_path,
+    when present) or None. Raises :class:`DbtConfigError` only on a
+    multi-compute conflict (different ``?o=`` values on http_paths that
+    share a connection). Other smells are emitted as warnings.
+    """
+    # SPOG can't activate on pre-SPOG deps; skip the matrix entirely.
+    if not (connector_supports_spog() and sdk_supports_workspace_id()):
+        return None
+
+    http_paths_list = list(http_paths)
+    extracted: set[str] = {
+        ws for p in http_paths_list if (ws := extract_workspace_id(p)) is not None
+    }
+
+    if len(extracted) > 1:
+        raise DbtConfigError(
+            f"Found conflicting workspace-id values across http_paths "
+            f"for host {host!r}: {sorted(extracted)}. Every compute used by "
+            f"this connection must agree on the workspace id."
+        )
+
+    workspace_id: Optional[str] = next(iter(extracted), None)
+    metadata = probe.probe_host(host)
+
+    if metadata.host_type == "unified" and workspace_id is None:
+        logger.warning(
+            f"Host {host!r} looks like a SPOG (unified) workspace but `http_path` "
+            f"does not include `?o=<workspace-id>`. Requests may fail to route. "
+            f"To opt into SPOG, set `http_path: /sql/1.0/warehouses/<id>?o=<workspace-id>` "
+            f"(workspace id is visible in the JDBC/ODBC connection-details panel)."
+        )
+    elif (
+        metadata.host_type is not None
+        and metadata.host_type != "unified"
+        and workspace_id is not None
+    ):
+        logger.warning(
+            f"`http_path` contains `?o={workspace_id}` but host {host!r} is not "
+            f"a SPOG workspace (host_type={metadata.host_type!r}). The `?o=` is "
+            f"likely ignored by the routing layer; remove it or use the SPOG "
+            f"hostname for this workspace if SPOG routing is intended."
+        )
+
+    return workspace_id
diff --git a/dbt/adapters/databricks/spog/extract.py b/dbt/adapters/databricks/spog/extract.py
new file mode 100644
index 000000000..fbb3c4503
--- /dev/null
+++ b/dbt/adapters/databricks/spog/extract.py
@@ -0,0 +1,12 @@
+"""Parse the SPOG workspace id out of a Databricks http_path."""
+
+from typing import Optional
+from urllib.parse import parse_qs
+
+
+def extract_workspace_id(http_path: Optional[str]) -> Optional[str]:
+    """Return the ``?o=<id>`` value from http_path, or None."""
+    if not http_path or "?" not in http_path:
+        return None
+    query = http_path.split("?", 1)[1]
+    return parse_qs(query).get("o", [None])[0]
diff --git a/dbt/adapters/databricks/spog/probe.py b/dbt/adapters/databricks/spog/probe.py
new file mode 100644
index 000000000..dc05c2af5
--- /dev/null
+++ b/dbt/adapters/databricks/spog/probe.py
@@ -0,0 +1,55 @@
+"""Discovery probe for /.well-known/databricks-config.
+
+Lets the adapter classify a host as SPOG (host_type='unified') or legacy.
+Probe is one-shot per host per process; result is cached.
+"""
+
+import random
+import time
+from dataclasses import dataclass
+from functools import cache
+from typing import Optional
+
+import requests
+
+from dbt.adapters.databricks.logging import logger
+
+
+@dataclass(frozen=True)
+class HostMetadata:
+    """Subset of /.well-known/databricks-config the adapter consumes."""
+
+    host_type: Optional[str]
+    account_id: Optional[str] = None
+
+
+@cache
+def probe_host(host: str) -> HostMetadata:
+    """Probe https://{host}/.well-known/databricks-config. Cached per host.
+
+    Retries up to 3 attempts with exponential backoff (~0.5s, 1s with jitter).
+    On exhaustion, logs a WARN and returns HostMetadata(host_type=None) so the
+    caller falls back to the legacy code path. Failure is never fatal.
+    """
+    url = f"https://{host}/.well-known/databricks-config"
+    last_exc: Optional[Exception] = None
+    for attempt in range(3):
+        try:
+            resp = requests.get(url, timeout=5)
+            resp.raise_for_status()
+            body = resp.json()
+            return HostMetadata(
+                host_type=body.get("host_type"),
+                account_id=body.get("account_id"),
+            )
+        except (requests.RequestException, ValueError) as e:
+            last_exc = e
+            if attempt < 2:
+                time.sleep((0.5 * (2**attempt)) + random.random() * 0.25)
+
+    logger.info(
+        f"SPOG discovery probe to {url!r} failed after 3 attempts (last error: {last_exc}). "
+        f"Proceeding as a non-SPOG host. If {host} is a SPOG (unified) workspace, routing "
+        f"errors may follow; verify network reachability to /.well-known/databricks-config."
+    )
+    return HostMetadata(host_type=None)
diff --git a/tests/functional/adapter/spog/__init__.py b/tests/functional/adapter/spog/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/functional/adapter/spog/conftest.py b/tests/functional/adapter/spog/conftest.py
new file mode 100644
index 000000000..39ced453a
--- /dev/null
+++ b/tests/functional/adapter/spog/conftest.py
@@ -0,0 +1,100 @@
+import os
+from copy import deepcopy
+from urllib.parse import parse_qsl, urlencode
+
+import pytest
+
+from dbt.adapters.databricks.spog.capabilities import (
+    connector_supports_spog,
+    sdk_supports_workspace_id,
+)
+from tests.profiles import get_databricks_cluster_target
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _require_spog_test_env():
+    if not (connector_supports_spog() and sdk_supports_workspace_id()):
+        pytest.skip(
+            "SPOG functional tests require databricks-sdk and databricks-sql-connector "
+            "with SPOG support installed."
+        )
+    if os.getenv("TEST_PECO_SPOG_NATIVE") == "1":
+        pytest.skip("SPOG override tests are redundant on a SPOG-native profile.")
+
+
+def _workspace_id() -> str:
+    """Return the SPOG workspace ID to point tests at."""
+    workspace_id = os.getenv("TEST_PECO_SPOG_WORKSPACE_ID")
+    if not workspace_id:
+        raise RuntimeError("SPOG functional tests require TEST_PECO_SPOG_WORKSPACE_ID to be set.")
+    return workspace_id
+
+
+def _spog_host() -> str:
+    """Return the SPOG host to point tests at."""
+    host = os.getenv("TEST_PECO_SPOG_HOST")
+    if not host:
+        raise RuntimeError("SPOG functional tests require TEST_PECO_SPOG_HOST to be set.")
+    return host
+
+
+def _with_workspace_id(http_path: str | None, workspace_id: str) -> str:
+    if not http_path:
+        raise RuntimeError("SPOG functional tests require an http_path.")
+
+    path, _, query = http_path.partition("?")
+    params = dict(parse_qsl(query, keep_blank_values=True))
+    params["o"] = workspace_id
+    return f"{path}?{urlencode(params)}"
+
+
+def _spog_target(profile_type: str, workspace_id: str) -> dict:
+    target = get_databricks_cluster_target(profile_type)
+    target["host"] = _spog_host()
+    target["http_path"] = _with_workspace_id(target.get("http_path"), workspace_id)
+    return target
+
+
+@pytest.fixture(scope="session")
+def dbt_profile_target(request):
+    profile_type = request.config.getoption("--profile")
+    workspace_id = _workspace_id()
+    return _spog_target(profile_type, workspace_id)
+
+
+@pytest.fixture(scope="class")
+def dbt_profile_data(unique_schema, dbt_profile_target, profiles_config_update):
+    workspace_id = _workspace_id()
+    target = deepcopy(dbt_profile_target)
+    target["schema"] = unique_schema
+
+    target["compute"] = {
+        "alternate_uc_cluster": {
+            "http_path": _spog_target("databricks_uc_cluster", workspace_id)["http_path"]
+        }
+    }
+
+    profile = {
+        "test": {
+            "outputs": {
+                "default": target,
+            },
+            "target": "default",
+        },
+    }
+
+    alternate_warehouse = deepcopy(target)
+    alternate_warehouse["compute"] = {
+        "alternate_warehouse": {"http_path": target["http_path"]},
+        "alternate_warehouse2": {"http_path": target["http_path"]},
+        "alternate_warehouse3": {"http_path": target["http_path"]},
+    }
+    profile["test"]["outputs"]["alternate_warehouse"] = alternate_warehouse
+
+    idle_sessions = deepcopy(alternate_warehouse)
+    idle_sessions["connect_max_idle"] = 1
+    profile["test"]["outputs"]["idle_sessions"] = idle_sessions
+
+    if profiles_config_update:
+        profile.update(profiles_config_update)
+    return profile
diff --git a/tests/functional/adapter/spog/fixtures.py b/tests/functional/adapter/spog/fixtures.py
new file mode 100644
index 000000000..1149f9168
--- /dev/null
+++ b/tests/functional/adapter/spog/fixtures.py
@@ -0,0 +1,10 @@
+spog_smoke_model = """
+select 1 as id
+"""
+
+
+spog_named_compute_model = """
+{{ config(databricks_compute='alternate_uc_cluster') }}
+
+select 1 as id
+"""
diff --git a/tests/functional/adapter/spog/test_spog_connection_open.py b/tests/functional/adapter/spog/test_spog_connection_open.py
new file mode 100644
index 000000000..ba538cae3
--- /dev/null
+++ b/tests/functional/adapter/spog/test_spog_connection_open.py
@@ -0,0 +1,42 @@
+from unittest import mock
+
+import pytest
+from dbt.tests.util import run_dbt
+
+from tests.functional.adapter.spog import fixtures
+
+
+class TestSpogConnectionOpen:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"spog_smoke_model.sql": fixtures.spog_smoke_model}
+
+    def test_spog_run_opens_connection(self, project):
+        run_dbt(["run", "--select", "spog_smoke_model"], expect_pass=True)
+
+
+class TestSpogProbeFailureFallback:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"spog_smoke_model.sql": fixtures.spog_smoke_model}
+
+    def test_probe_failure_run_succeeds(self, project):
+        from dbt.adapters.databricks.spog import probe
+        from dbt.adapters.databricks.spog.probe import HostMetadata
+
+        probe.probe_host.cache_clear()
+        with mock.patch(
+            "dbt.adapters.databricks.spog.probe.probe_host",
+            return_value=HostMetadata(host_type=None),
+        ):
+            run_dbt(["run", "--select", "spog_smoke_model"], expect_pass=True)
+
+
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestSpogNamedCompute:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"spog_named_compute_model.sql": fixtures.spog_named_compute_model}
+
+    def test_spog_named_compute_same_workspace_succeeds(self, project):
+        run_dbt(["run", "--select", "spog_named_compute_model"], expect_pass=True)
diff --git a/tests/functional/adapter/spog/test_spog_debug.py b/tests/functional/adapter/spog/test_spog_debug.py
new file mode 100644
index 000000000..b70f8c870
--- /dev/null
+++ b/tests/functional/adapter/spog/test_spog_debug.py
@@ -0,0 +1,6 @@
+from dbt.tests.util import run_dbt
+
+
+class TestSpogDebugOutput:
+    def test_dbt_debug_reports_spog(self, project):
+        run_dbt(["debug"], expect_pass=True)
diff --git a/tests/unit/python/test_python_helpers.py b/tests/unit/python/test_python_helpers.py
index 8c93bf20e..f2e83f4c9 100644
--- a/tests/unit/python/test_python_helpers.py
+++ b/tests/unit/python/test_python_helpers.py
@@ -23,6 +23,7 @@ def credentials():
     c = Mock()
     c.get_all_http_headers.return_value = {}
     c.connection_parameters = {}
+    c.http_path = "/sql/1.0/warehouses/abc"
     return c
 
 
diff --git a/tests/unit/spog/__init__.py b/tests/unit/spog/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/unit/spog/conftest.py b/tests/unit/spog/conftest.py
new file mode 100644
index 000000000..706ca1400
--- /dev/null
+++ b/tests/unit/spog/conftest.py
@@ -0,0 +1,18 @@
+from unittest import mock
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _guard_unmocked_requests_get():
+    """Force any HTTP call inside probe.py to be explicitly mocked at the
+    test level. Catches the "forgot to mock" failure mode with a loud
+    AssertionError rather than letting it silently hit the network."""
+    with mock.patch(
+        "dbt.adapters.databricks.spog.probe.requests.get",
+        side_effect=AssertionError(
+            "Unmocked requests.get in a SPOG unit test. Patch "
+            "'dbt.adapters.databricks.spog.probe.requests.get' inside your test."
+        ),
+    ):
+        yield
diff --git a/tests/unit/spog/test_capabilities.py b/tests/unit/spog/test_capabilities.py
new file mode 100644
index 000000000..9faea59c0
--- /dev/null
+++ b/tests/unit/spog/test_capabilities.py
@@ -0,0 +1,52 @@
+from importlib.metadata import PackageNotFoundError
+from unittest import mock
+
+import pytest
+
+from dbt.adapters.databricks.spog import capabilities
+
+
+@pytest.fixture(autouse=True)
+def clear_caches():
+    """Clear @cache decorators between tests so each test sees a fresh state."""
+    capabilities.connector_supports_spog.cache_clear()
+    capabilities.sdk_supports_workspace_id.cache_clear()
+    yield
+    capabilities.connector_supports_spog.cache_clear()
+    capabilities.sdk_supports_workspace_id.cache_clear()
+
+
+class TestConnectorSupportsSpog:
+    @pytest.mark.parametrize(
+        ("connector_version", "expected"),
+        [
+            ("4.2.5", False),
+            ("4.2.6", True),
+        ],
+    )
+    def test_checks_connector_version_floor(self, connector_version, expected):
+        with mock.patch(
+            "dbt.adapters.databricks.spog.capabilities._pkg_version",
+            return_value=connector_version,
+        ):
+            assert capabilities.connector_supports_spog() is expected
+
+    def test_missing_package_returns_false(self):
+        with mock.patch(
+            "dbt.adapters.databricks.spog.capabilities._pkg_version",
+            side_effect=PackageNotFoundError("databricks-sql-connector"),
+        ):
+            assert capabilities.connector_supports_spog() is False
+
+
+class TestSdkSupportsWorkspaceId:
+    @pytest.mark.parametrize(
+        ("config_cls", "expected"),
+        [
+            (type("ConfigWithWorkspaceId", (), {"workspace_id": None}), True),
+            (type("ConfigWithoutWorkspaceId", (), {"host": None}), False),
+        ],
+    )
+    def test_checks_config_workspace_id_attribute(self, config_cls, expected):
+        with mock.patch("dbt.adapters.databricks.spog.capabilities.Config", config_cls):
+            assert capabilities.sdk_supports_workspace_id() is expected
diff --git a/tests/unit/spog/test_decision.py b/tests/unit/spog/test_decision.py
new file mode 100644
index 000000000..0b683a9da
--- /dev/null
+++ b/tests/unit/spog/test_decision.py
@@ -0,0 +1,167 @@
+from contextlib import ExitStack
+from unittest import mock
+
+import pytest
+from dbt_common.exceptions import DbtConfigError
+
+from dbt.adapters.databricks.spog import decision
+from dbt.adapters.databricks.spog.probe import HostMetadata
+
+
+@pytest.fixture
+def patch_decision():
+    """Patch capability + probe modules to fixed responses for one test.
+
+    Yields a function: patch_decision(host_type=..., connector_ok=True, sdk_ok=True).
+    All patches auto-unwind at test teardown.
+    """
+    with ExitStack() as stack:
+
+        def _apply(*, host_type, connector_ok=True, sdk_ok=True):
+            stack.enter_context(
+                mock.patch(
+                    "dbt.adapters.databricks.spog.probe.probe_host",
+                    return_value=HostMetadata(host_type=host_type),
+                )
+            )
+            stack.enter_context(
+                mock.patch(
+                    "dbt.adapters.databricks.spog.decision.connector_supports_spog",
+                    return_value=connector_ok,
+                )
+            )
+            stack.enter_context(
+                mock.patch(
+                    "dbt.adapters.databricks.spog.decision.sdk_supports_workspace_id",
+                    return_value=sdk_ok,
+                )
+            )
+
+        yield _apply
+
+
+class TestCheckSpogPreconditions:
+    def test_spog_happy_path(self, patch_decision):
+        patch_decision(host_type="unified")
+        ws_id = decision.check_spog_preconditions(
+            host="spog.example.com",
+            http_paths=["/sql/1.0/warehouses/abc?o=64"],
+        )
+        assert ws_id == "64"
+
+    def test_spog_missing_o_warns_and_returns_none(self, patch_decision):
+        patch_decision(host_type="unified")
+        with mock.patch("dbt.adapters.databricks.spog.decision.logger") as mock_logger:
+            ws_id = decision.check_spog_preconditions(
+                host="spog.example.com",
+                http_paths=["/sql/1.0/warehouses/abc"],
+            )
+        assert ws_id is None
+        mock_logger.warning.assert_called_once()
+
+    def test_non_spog_with_explicit_o_warns_and_returns_id(self, patch_decision):
+        patch_decision(host_type="workspace")
+        with mock.patch("dbt.adapters.databricks.spog.decision.logger") as mock_logger:
+            ws_id = decision.check_spog_preconditions(
+                host="legacy.example.com",
+                http_paths=["/sql/1.0/warehouses/abc?o=64"],
+            )
+        assert ws_id == "64"
+        mock_logger.warning.assert_called_once()
+
+    @pytest.mark.parametrize(
+        ("connector_ok", "sdk_ok"),
+        [(True, False), (False, True)],
+        ids=["sdk-too-old", "connector-too-old"],
+    )
+    def test_pre_spog_deps_short_circuit_no_probe(self, connector_ok, sdk_ok):
+        """When either dep is below the SPOG floor, return None without probing.
+        SPOG can't activate on pre-SPOG deps anyway."""
+        with (
+            mock.patch(
+                "dbt.adapters.databricks.spog.decision.connector_supports_spog",
+                return_value=connector_ok,
+            ),
+            mock.patch(
+                "dbt.adapters.databricks.spog.decision.sdk_supports_workspace_id",
+                return_value=sdk_ok,
+            ),
+            mock.patch("dbt.adapters.databricks.spog.probe.probe_host") as probe,
+        ):
+            result = decision.check_spog_preconditions(
+                host="spog.example.com",
+                http_paths=["/sql/1.0/warehouses/abc?o=64"],
+            )
+        assert result is None
+        probe.assert_not_called()
+
+    def test_legacy_cluster_path_no_o_returns_none_no_warn(self, patch_decision):
+        """Cluster path /o/<id>/ is *not* a SPOG declaration. On a legacy host
+        with no ?o=, there's nothing for us to mirror and nothing to warn
+        about — `?o=` is the only opt-in marker."""
+        patch_decision(host_type="workspace")
+        with mock.patch("dbt.adapters.databricks.spog.decision.logger") as mock_logger:
+            ws_id = decision.check_spog_preconditions(
+                host="legacy.example.com",
+                http_paths=[
+                    "/sql/1.0/warehouses/default",
+                    "/sql/protocolv1/o/6436897454825492/0605-cluster",
+                ],
+            )
+        assert ws_id is None
+        mock_logger.warning.assert_not_called()
+
+    def test_non_spog_legacy_returns_none(self, patch_decision):
+        patch_decision(host_type="workspace")
+        assert (
+            decision.check_spog_preconditions(
+                host="legacy.example.com",
+                http_paths=["/sql/1.0/warehouses/abc"],
+            )
+            is None
+        )
+
+    def test_probe_failure_returns_extracted_id(self, patch_decision):
+        """Probe failure is non-fatal; surface whatever the user explicitly
+        declared via ?o= and let downstream handle routing."""
+        patch_decision(host_type=None)
+        ws_id = decision.check_spog_preconditions(
+            host="flaky.example.com",
+            http_paths=["/sql/1.0/warehouses/abc?o=64"],
+        )
+        assert ws_id == "64"
+
+    def test_probe_failure_without_o_returns_none(self, patch_decision):
+        patch_decision(host_type=None)
+        assert (
+            decision.check_spog_preconditions(
+                host="flaky.example.com",
+                http_paths=["/sql/1.0/warehouses/abc"],
+            )
+            is None
+        )
+
+    def test_multi_compute_consistent_returns_id(self, patch_decision):
+        patch_decision(host_type="unified")
+        ws_id = decision.check_spog_preconditions(
+            host="spog.example.com",
+            http_paths=[
+                "/sql/1.0/warehouses/a?o=64",
+                "/sql/1.0/warehouses/b?o=64",
+            ],
+        )
+        assert ws_id == "64"
+
+    def test_multi_compute_conflicting_raises(self, patch_decision):
+        """The only hard raise: different ?o= values on http_paths that
+        share a connection. This is unambiguously broken regardless of
+        host type, so we fail fast instead of warning."""
+        patch_decision(host_type="unified")
+        with pytest.raises(DbtConfigError, match=r"conflicting"):
+            decision.check_spog_preconditions(
+                host="spog.example.com",
+                http_paths=[
+                    "/sql/1.0/warehouses/a?o=64",
+                    "/sql/1.0/warehouses/b?o=99",
+                ],
+            )
diff --git a/tests/unit/spog/test_extract.py b/tests/unit/spog/test_extract.py
new file mode 100644
index 000000000..5ff54d5f7
--- /dev/null
+++ b/tests/unit/spog/test_extract.py
@@ -0,0 +1,51 @@
+from dbt.adapters.databricks.spog.extract import extract_workspace_id
+
+
+class TestExtractWorkspaceId:
+    """`?o=<id>` is the only canonical SPOG marker. The cluster path
+    `/o/<id>/` segment is *not* a SPOG declaration — the connector's
+    `_extract_spog_headers` only reads `?o=`, so we mirror that contract."""
+
+    def test_warehouse_path_with_o_param(self):
+        assert (
+            extract_workspace_id("/sql/1.0/warehouses/abc123?o=6436897454825492")
+            == "6436897454825492"
+        )
+
+    def test_cluster_path_with_o_param(self):
+        assert (
+            extract_workspace_id(
+                "/sql/protocolv1/o/2548836972759138/0605-142813-rf81cyrh?o=6436897454825492"
+            )
+            == "6436897454825492"
+        )
+
+    def test_no_query_string(self):
+        assert extract_workspace_id("/sql/1.0/warehouses/abc123") is None
+
+    def test_query_string_no_o_param(self):
+        assert extract_workspace_id("/sql/1.0/warehouses/abc123?other=value") is None
+
+    def test_multiple_query_params(self):
+        assert extract_workspace_id("/sql/1.0/warehouses/abc123?o=12345&ts=1") == "12345"
+
+    def test_o_param_not_first(self):
+        assert extract_workspace_id("/sql/1.0/warehouses/abc123?ts=1&o=12345") == "12345"
+
+    def test_empty_string(self):
+        assert extract_workspace_id("") is None
+
+    def test_none_input(self):
+        assert extract_workspace_id(None) is None
+
+    def test_duplicate_o_params_returns_first(self):
+        # parse_qs returns ['a', 'b']; we take the first to be deterministic
+        assert extract_workspace_id("/path?o=a&o=b") == "a"
+
+    def test_cluster_path_without_o_returns_none(self):
+        """Cluster paths encode workspace id in `/o/<id>/` but that is *not*
+        a SPOG opt-in — only `?o=` is. Users on cluster paths must add
+        `?o=<workspace-id>` to declare SPOG intent."""
+        assert (
+            extract_workspace_id("/sql/protocolv1/o/6436897454825492/1214-195625-oc3mas1h") is None
+        )
diff --git a/tests/unit/spog/test_probe.py b/tests/unit/spog/test_probe.py
new file mode 100644
index 000000000..bf36bd322
--- /dev/null
+++ b/tests/unit/spog/test_probe.py
@@ -0,0 +1,75 @@
+from unittest import mock
+
+import pytest
+import requests
+
+from dbt.adapters.databricks.spog import probe
+from dbt.adapters.databricks.spog.probe import HostMetadata
+
+
+@pytest.fixture(autouse=True)
+def clear_probe_cache():
+    probe.probe_host.cache_clear()
+    yield
+    probe.probe_host.cache_clear()
+
+
+def _mock_response(json_body, status_code=200):
+    resp = mock.Mock(spec=requests.Response)
+    resp.status_code = status_code
+    resp.json.return_value = json_body
+    resp.raise_for_status = mock.Mock()
+    return resp
+
+
+class TestProbeHostHappyPath:
+    def test_unified_host(self):
+        with mock.patch(
+            "dbt.adapters.databricks.spog.probe.requests.get",
+            return_value=_mock_response({"host_type": "unified", "account_id": "acct-123"}),
+        ) as m:
+            result = probe.probe_host("spog.example.com")
+        assert result == HostMetadata(host_type="unified", account_id="acct-123")
+        m.assert_called_once_with(
+            "https://spog.example.com/.well-known/databricks-config",
+            timeout=5,
+        )
+
+    def test_result_cached_per_host(self):
+        with mock.patch(
+            "dbt.adapters.databricks.spog.probe.requests.get",
+            return_value=_mock_response({"host_type": "unified"}),
+        ) as m:
+            probe.probe_host("spog.example.com")
+            probe.probe_host("spog.example.com")
+        assert m.call_count == 1
+
+
+class TestProbeHostRetry:
+    def test_retries_three_times_on_failure(self):
+        with (
+            mock.patch(
+                "dbt.adapters.databricks.spog.probe.requests.get",
+                side_effect=requests.ConnectionError("network down"),
+            ) as m,
+            mock.patch("dbt.adapters.databricks.spog.probe.time.sleep"),
+            mock.patch("dbt.adapters.databricks.spog.probe.logger"),
+        ):
+            result = probe.probe_host("flaky.example.com")
+        assert m.call_count == 3
+        assert result == HostMetadata(host_type=None)
+
+    def test_json_decode_error_treated_as_failure(self):
+        bad_resp = mock.Mock(spec=requests.Response)
+        bad_resp.status_code = 200
+        bad_resp.raise_for_status = mock.Mock()
+        bad_resp.json.side_effect = ValueError("not json")
+        with (
+            mock.patch(
+                "dbt.adapters.databricks.spog.probe.requests.get",
+                return_value=bad_resp,
+            ),
+            mock.patch("dbt.adapters.databricks.spog.probe.time.sleep"),
+        ):
+            result = probe.probe_host("badjson.example.com")
+        assert result == HostMetadata(host_type=None)
diff --git a/tests/unit/test_adapter.py b/tests/unit/test_adapter.py
index 0c2b5ca80..83cb96152 100644
--- a/tests/unit/test_adapter.py
+++ b/tests/unit/test_adapter.py
@@ -113,6 +113,17 @@ def _get_config(
 
 
 class TestDatabricksAdapter(DatabricksAdapterBase):
+    @pytest.fixture(autouse=True)
+    def _stub_spog_probe(self):
+        """Stub `probe_host` to keep connection-open tests offline."""
+        from dbt.adapters.databricks.spog.probe import HostMetadata
+
+        with patch(
+            "dbt.adapters.databricks.spog.probe.probe_host",
+            return_value=HostMetadata(host_type=None),
+        ):
+            yield
+
     def test_two_catalog_settings(self):
         with pytest.raises(DbtConfigError) as excinfo:
             self._get_config(
@@ -1835,3 +1846,84 @@ def record(event_msg):
             "get_behavior_flag_no_warn must not fire BehaviorChangeEvent for "
             f"use_materialization_v2 (fired {len(caught)} times)"
         )
+
+
+class TestDebugEmitSpogBlock:
+    """Verify debug_emit_spog_block surfaces SPOG status diagnostics."""
+
+    def test_spog_host_with_capable_deps(self):
+        from dbt.adapters.databricks.impl import DatabricksAdapter
+        from dbt.adapters.databricks.spog.probe import HostMetadata
+
+        with (
+            patch(
+                "dbt.adapters.databricks.spog.probe.probe_host",
+                return_value=HostMetadata(host_type="unified", account_id="acct"),
+            ),
+            patch("dbt.adapters.databricks.impl.connector_supports_spog", return_value=True),
+            patch("dbt.adapters.databricks.impl.sdk_supports_workspace_id", return_value=True),
+            patch("dbt.adapters.databricks.impl.logger") as mock_logger,
+        ):
+            DatabricksAdapter.debug_emit_spog_block(
+                host="spog.example.com",
+                http_path="/sql/1.0/warehouses/abc?o=6436897454825492",
+            )
+
+        joined = " ".join(call.args[0] for call in mock_logger.info.call_args_list)
+        assert "SPOG host" in joined
+        assert "yes" in joined
+        assert "6436897454825492" in joined
+        assert "supported" in joined
+        assert "NOT supported" not in joined
+
+    def test_non_spog_host(self):
+        from dbt.adapters.databricks.impl import DatabricksAdapter
+        from dbt.adapters.databricks.spog.probe import HostMetadata
+
+        with (
+            patch(
+                "dbt.adapters.databricks.spog.probe.probe_host",
+                return_value=HostMetadata(host_type="workspace"),
+            ),
+            patch("dbt.adapters.databricks.impl.connector_supports_spog", return_value=True),
+            patch("dbt.adapters.databricks.impl.sdk_supports_workspace_id", return_value=True),
+            patch("dbt.adapters.databricks.impl.logger") as mock_logger,
+        ):
+            DatabricksAdapter.debug_emit_spog_block(
+                host="legacy.example.com",
+                http_path="/sql/1.0/warehouses/abc",
+            )
+        joined = " ".join(call.args[0] for call in mock_logger.info.call_args_list)
+        # 'no' for the SPOG host line
+        assert "SPOG host" in joined and "no" in joined
+
+    def test_spog_host_with_old_connector_flags_unsupported(self):
+        from dbt.adapters.databricks.impl import DatabricksAdapter
+        from dbt.adapters.databricks.spog.probe import HostMetadata
+
+        with (
+            patch(
+                "dbt.adapters.databricks.spog.probe.probe_host",
+                return_value=HostMetadata(host_type="unified"),
+            ),
+            patch("dbt.adapters.databricks.impl.connector_supports_spog", return_value=False),
+            patch("dbt.adapters.databricks.impl.sdk_supports_workspace_id", return_value=True),
+            patch("dbt.adapters.databricks.impl.logger") as mock_logger,
+        ):
+            DatabricksAdapter.debug_emit_spog_block(
+                host="spog.example.com",
+                http_path="/sql/1.0/warehouses/abc?o=64",
+            )
+        joined = " ".join(call.args[0] for call in mock_logger.info.call_args_list)
+        assert "NOT supported" in joined
+
+    def test_no_op_on_empty_host(self):
+        from dbt.adapters.databricks.impl import DatabricksAdapter
+
+        with (
+            patch("dbt.adapters.databricks.spog.probe.probe_host") as mock_probe,
+            patch("dbt.adapters.databricks.impl.logger") as mock_logger,
+        ):
+            DatabricksAdapter.debug_emit_spog_block(host="", http_path="")
+        mock_probe.assert_not_called()
+        mock_logger.info.assert_not_called()
diff --git a/tests/unit/test_auth.py b/tests/unit/test_auth.py
index e671847d2..9b886739f 100644
--- a/tests/unit/test_auth.py
+++ b/tests/unit/test_auth.py
@@ -7,7 +7,10 @@
 import pytest
 from databricks.sdk import config as _sdk_config
 
-from dbt.adapters.databricks.credentials import DatabricksCredentials
+from dbt.adapters.databricks.credentials import (
+    DatabricksCredentialManager,
+    DatabricksCredentials,
+)
 
 # databricks-sdk >=0.103 added a network probe to Config(); on older SDKs
 # this symbol doesn't exist and the lazy `_ensure_config` path is unnecessary.
@@ -318,3 +321,145 @@ def delete_password(self, servicename, username):
             return None
 
         os.remove(file_path)
+
+
+class TestCredentialManagerWorkspaceId:
+    """Verify workspace_id is extracted from http_path and plumbed into Config."""
+
+    def _creds(self, http_path: str, token: str = "dapi-fake") -> DatabricksCredentials:
+        return DatabricksCredentials(
+            host="spog.example.com",
+            http_path=http_path,
+            token=token,
+            database="main",
+            schema="default",
+        )
+
+    def test_workspace_id_extracted_when_present(self):
+        creds = self._creds("/sql/1.0/warehouses/abc?o=6436897454825492")
+        mgr = DatabricksCredentialManager.create_from(creds)
+        assert mgr.workspace_id == "6436897454825492"
+
+    def test_workspace_id_none_when_absent(self):
+        creds = self._creds("/sql/1.0/warehouses/abc")
+        mgr = DatabricksCredentialManager.create_from(creds)
+        assert mgr.workspace_id is None
+
+    def test_pat_config_receives_workspace_id_when_supported(self):
+        creds = self._creds("/sql/1.0/warehouses/abc?o=64")
+        with (
+            mock.patch(
+                "dbt.adapters.databricks.credentials.sdk_supports_workspace_id",
+                return_value=True,
+            ),
+            mock.patch("dbt.adapters.databricks.credentials.Config") as cfg,
+        ):
+            mgr = DatabricksCredentialManager.create_from(creds)
+            mgr.authenticate_with_pat()
+        kwargs = cfg.call_args.kwargs
+        assert kwargs["host"] == "spog.example.com"
+        assert kwargs["token"] == "dapi-fake"
+        assert kwargs["workspace_id"] == "64"
+
+    def test_pat_config_no_workspace_id_when_unsupported(self):
+        creds = self._creds("/sql/1.0/warehouses/abc?o=64")
+        with (
+            mock.patch(
+                "dbt.adapters.databricks.credentials.sdk_supports_workspace_id",
+                return_value=False,
+            ),
+            mock.patch("dbt.adapters.databricks.credentials.Config") as cfg,
+        ):
+            mgr = DatabricksCredentialManager.create_from(creds)
+            mgr.authenticate_with_pat()
+        kwargs = cfg.call_args.kwargs
+        assert kwargs["host"] == "spog.example.com"
+        assert kwargs["token"] == "dapi-fake"
+        assert "workspace_id" not in kwargs
+
+    def test_pat_config_no_workspace_id_when_no_o_param(self):
+        creds = self._creds("/sql/1.0/warehouses/abc")  # no ?o=
+        with (
+            mock.patch(
+                "dbt.adapters.databricks.credentials.sdk_supports_workspace_id",
+                return_value=True,
+            ),
+            mock.patch("dbt.adapters.databricks.credentials.Config") as cfg,
+        ):
+            mgr = DatabricksCredentialManager.create_from(creds)
+            mgr.authenticate_with_pat()
+        kwargs = cfg.call_args.kwargs
+        assert kwargs["host"] == "spog.example.com"
+        assert kwargs["token"] == "dapi-fake"
+        assert "workspace_id" not in kwargs
+
+    def test_oauth_m2m_config_receives_workspace_id(self):
+        creds = DatabricksCredentials(
+            host="spog.example.com",
+            http_path="/sql/1.0/warehouses/abc?o=64",
+            client_id="cid",
+            client_secret="csec",
+            database="main",
+            schema="default",
+        )
+        with (
+            mock.patch(
+                "dbt.adapters.databricks.credentials.sdk_supports_workspace_id",
+                return_value=True,
+            ),
+            mock.patch("dbt.adapters.databricks.credentials.Config") as cfg,
+        ):
+            mgr = DatabricksCredentialManager.create_from(creds)
+            mgr.authenticate_with_oauth_m2m()
+        kwargs = cfg.call_args.kwargs
+        assert kwargs["host"] == "spog.example.com"
+        assert kwargs["client_id"] == "cid"
+        assert kwargs["client_secret"] == "csec"
+        assert kwargs["workspace_id"] == "64"
+        assert kwargs["auth_type"] == "oauth-m2m"
+
+    def test_azure_client_secret_config_receives_workspace_id(self):
+        creds = DatabricksCredentials(
+            host="spog.example.com",
+            http_path="/sql/1.0/warehouses/abc?o=64",
+            azure_client_id="az-cid",
+            azure_client_secret="az-csec",
+            database="main",
+            schema="default",
+        )
+        with (
+            mock.patch(
+                "dbt.adapters.databricks.credentials.sdk_supports_workspace_id",
+                return_value=True,
+            ),
+            mock.patch("dbt.adapters.databricks.credentials.Config") as cfg,
+        ):
+            mgr = DatabricksCredentialManager.create_from(creds)
+            mgr.authenticate_with_azure_client_secret()
+        kwargs = cfg.call_args.kwargs
+        assert kwargs["host"] == "spog.example.com"
+        assert kwargs["workspace_id"] == "64"
+        assert kwargs["azure_client_id"] == "az-cid"
+
+    def test_external_browser_config_receives_workspace_id(self):
+        with (
+            mock.patch(
+                "dbt.adapters.databricks.credentials.sdk_supports_workspace_id",
+                return_value=True,
+            ),
+            mock.patch("dbt.adapters.databricks.credentials.Config") as cfg,
+        ):
+            creds = DatabricksCredentials(
+                host="spog.example.com",
+                http_path="/sql/1.0/warehouses/abc?o=64",
+                client_id="cid",
+                database="main",
+                schema="default",
+            )
+            mgr = DatabricksCredentialManager.create_from(creds)
+            mgr.authenticate_with_external_browser()
+        kwargs = cfg.call_args.kwargs
+        assert kwargs["host"] == "spog.example.com"
+        assert kwargs["client_id"] == "cid"
+        assert kwargs["workspace_id"] == "64"
+        assert kwargs["auth_type"] == "external-browser"
diff --git a/tests/unit/test_connection_manager.py b/tests/unit/test_connection_manager.py
index 536d6e04b..1561b7ebc 100644
--- a/tests/unit/test_connection_manager.py
+++ b/tests/unit/test_connection_manager.py
@@ -75,6 +75,8 @@ def test_open_calls_is_cluster_http_path_for_warehouse(
         mock_connection.credentials.connect_retries = 1
         mock_connection.credentials.connect_timeout = 10
         mock_connection.credentials.query_tags = None
+        mock_connection.credentials.compute = None  # default for real DatabricksCredentials
+        mock_connection.credentials.host = "example.cloud.databricks.com"
         mock_connection.http_path = "sql/protocolv1/o/abc123def456"
         mock_connection.credentials.authenticate.return_value = Mock()
         mock_connection._query_header_context = None
@@ -161,3 +163,50 @@ def test_retry_succeeds_after_transient_failure(self):
         assert caps is not None
         assert caps.dbr_version == (16, 2)
         assert caps.has_capability(DBRCapability.ICEBERG)
+
+
+class TestOpenSpogIntegration:
+    """Verify DatabricksConnectionManager.open() invokes check_spog_preconditions."""
+
+    def test_passes_all_http_paths_to_check(self):
+        captured = {}
+
+        def fake_check(*, host, http_paths):
+            captured["host"] = host
+            captured["http_paths"] = list(http_paths)
+            return None
+
+        with (
+            patch(
+                "dbt.adapters.databricks.connections.check_spog_preconditions",
+                side_effect=fake_check,
+            ),
+            patch("dbt.adapters.databricks.connections.DatabricksHandle.from_connection_args"),
+            patch(
+                "dbt.adapters.databricks.connections.SqlUtils.prepare_connection_arguments",
+                return_value={},
+            ),
+            patch(
+                "dbt.adapters.databricks.connections.QueryConfigUtils.get_merged_query_tags",
+                return_value={},
+            ),
+        ):
+            fake_conn = Mock()
+            fake_conn.state = "init"
+            fake_conn.credentials = Mock(spec=DatabricksCredentials)
+            fake_conn.credentials.host = "spog.example.com"
+            fake_conn.credentials.http_path = "/sql/1.0/warehouses/default?o=64"
+            fake_conn.credentials.compute = {"extra": {"http_path": "/sql/1.0/warehouses/alt?o=64"}}
+            fake_conn.credentials.connect_timeout = None
+            fake_conn.credentials.connect_retries = 1
+            fake_conn.credentials.retry_all = False
+            fake_conn.credentials.cluster_id = None
+            fake_conn.credentials.authenticate = Mock(return_value=Mock())
+            fake_conn.http_path = "/sql/1.0/warehouses/default?o=64"
+            fake_conn._query_header_context = None
+
+            DatabricksConnectionManager.open(fake_conn)
+
+        assert captured["host"] == "spog.example.com"
+        assert "/sql/1.0/warehouses/default?o=64" in captured["http_paths"]
+        assert "/sql/1.0/warehouses/alt?o=64" in captured["http_paths"]
diff --git a/tests/unit/test_credentials_cluster_id_regex.py b/tests/unit/test_credentials_cluster_id_regex.py
new file mode 100644
index 000000000..05699dfb7
--- /dev/null
+++ b/tests/unit/test_credentials_cluster_id_regex.py
@@ -0,0 +1,44 @@
+from dbt.adapters.databricks.credentials import DatabricksCredentials
+
+
+class TestExtractClusterId:
+    def test_legacy_cluster_path(self):
+        assert (
+            DatabricksCredentials.extract_cluster_id(
+                "sql/protocolv1/o/2548836972759138/0605-142813-rf81cyrh"
+            )
+            == "0605-142813-rf81cyrh"
+        )
+
+    def test_cluster_path_with_leading_slash(self):
+        assert (
+            DatabricksCredentials.extract_cluster_id(
+                "/sql/protocolv1/o/2548836972759138/0605-142813-rf81cyrh"
+            )
+            == "0605-142813-rf81cyrh"
+        )
+
+    def test_cluster_path_with_spog_query_param(self):
+        # ?o=<workspace-id> must NOT be captured into the cluster id
+        assert (
+            DatabricksCredentials.extract_cluster_id(
+                "/sql/protocolv1/o/2548836972759138/0605-142813-rf81cyrh?o=6436897454825492"
+            )
+            == "0605-142813-rf81cyrh"
+        )
+
+    def test_cluster_path_with_multiple_query_params(self):
+        assert (
+            DatabricksCredentials.extract_cluster_id(
+                "/sql/protocolv1/o/2548836972759138/0605-142813-rf81cyrh?o=64&ts=1"
+            )
+            == "0605-142813-rf81cyrh"
+        )
+
+    def test_warehouse_path_returns_none(self):
+        assert (
+            DatabricksCredentials.extract_cluster_id(
+                "/sql/1.0/warehouses/abc123?o=6436897454825492"
+            )
+            is None
+        )