diff --git a/eng/pipelines/pr-validation-pipeline.yml b/eng/pipelines/pr-validation-pipeline.yml
index 66db771d..373f8dd0 100644
--- a/eng/pipelines/pr-validation-pipeline.yml
+++ b/eng/pipelines/pr-validation-pipeline.yml
@@ -470,6 +470,9 @@ jobs:
     vmImage: 'macos-latest'
 
   variables:
+    # Apply to setup, benchmark consumers and final cleanup without changing
+    # the Docker client's global context.
+    DOCKER_CONTEXT: colima
     # pip's on-disk download/wheel cache. Persisted across runs by the Cache@2
     # task below so `pip install -r requirements.txt` reuses wheels instead of
     # re-downloading them on every run. Exposed to script steps as the
@@ -517,70 +520,39 @@ jobs:
 
   - script: |
       set -e
-      # Overlap the ENTIRE container-side setup (Colima VM boot -> image pull ->
-      # container start -> SQL readiness) with the CPU-bound Python setup (pip
-      # install + pybind build). The two chains are independent until pytest, so
-      # we run the container chain in the background and the build in the
-      # foreground, then wait for the container before continuing.
-      #
-      # On the Intel macOS-15 hosted runner the linux/amd64 SQL Server image runs
-      # natively (no emulation); the dominant container costs are the Colima VM
-      # boot (~2-4 min) and the ~2 GB image pull, both of which are hidden behind
-      # the build here.
-      #
-      # NOTE: this deliberately merges what used to be separate steps (start
-      # Colima, pull+start SQL, install Python deps, build pybind). Splitting them
-      # back out restores the previous per-step timing breakdown.
-      setup_sql() {
-        echo "[sql] Starting Colima VM..."
-        colima start --cpu 4 --memory 8 --disk 50
-        docker context use colima >/dev/null || true
-        docker version
-
-        echo "[sql] Pulling SQL Server image..."
-        docker pull "$(sqlServerImage)"
-
-        echo "[sql] Starting SQL Server container..."
-        docker run \
-          --name sqlserver \
-          -e ACCEPT_EULA=Y \
-          -e MSSQL_SA_PASSWORD="${DB_PASSWORD}" \
-          -p 1433:1433 \
-          -d "$(sqlServerImage)"
-
-        echo "[sql] Waiting for SQL Server to accept connections..."
-        for i in {1..30}; do
-          if docker exec sqlserver \
-            /opt/mssql-tools18/bin/sqlcmd \
-            -S localhost -U SA -P "$DB_PASSWORD" \
-            -C -Q "SELECT 1"; then
-            echo "[sql] SQL Server is ready."
-            return 0
+      # Only SQL lifecycle is retried. Colima starts once, overlapped with the
+      # unchanged dependency/native build chain. The helper bounds SQL work
+      # and redacts output before it reaches this job-private log.
+      SQL_LOG="$(Agent.TempDirectory)/sql-setup-$(System.JobId).log"
+      SQL_HELPER="$(Build.SourcesDirectory)/eng/scripts/setup_sql_container.py"
+      SQL_OWNER="$(Build.BuildId).$(System.JobId)"
+      SQL_PID=
+      finish_setup() {
+        status=$?
+        trap - EXIT INT TERM
+        if [ -n "$SQL_PID" ]; then
+          if kill -0 "$SQL_PID" 2>/dev/null; then
+            kill -TERM "$SQL_PID" || echo "[sql] Background setup already stopped"
           fi
-          sleep 2
-        done
-
-        # One last probe: the loop sleeps after its final attempt, so SQL can
-        # become reachable in that window and would otherwise be missed.
-        if docker exec sqlserver \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost -U SA -P "$DB_PASSWORD" \
-          -C -Q "SELECT 1"; then
-          echo "[sql] SQL Server is ready."
-          return 0
+          wait "$SQL_PID" || echo "[sql] Background setup stopped or failed"
         fi
-
-        # sqlcmd only reports that the client could not connect. When the
-        # container itself exited (bad password policy, EULA, OOM, port clash)
-        # the reason is in the container's own log, so surface it here.
-        echo "[sql] SQL Server did not become ready in time." >&2
-        echo "[sql] ---- docker logs sqlserver (last 200 lines) ----" >&2
-        docker logs --tail 200 sqlserver >&2 || true
-        return 1
+        if [ -f "$SQL_LOG" ]; then
+          cat "$SQL_LOG"
+          rm -f "$SQL_LOG"
+        fi
+        if [ "$status" -ne 0 ]; then
+          python3 "$SQL_HELPER" --name sqlserver --owner "$SQL_OWNER" --colima --cleanup \
+            || echo "[sql] Cleanup failed; final job cleanup will also run"
+        fi
+        exit "$status"
       }
+      trap finish_setup EXIT
+      trap 'exit 130' INT
+      trap 'exit 143' TERM
 
       echo "Starting container setup (Colima + SQL Server) in the background..."
-      setup_sql > /tmp/sql_setup.log 2>&1 &
+      python3 "$SQL_HELPER" --name sqlserver --owner "$SQL_OWNER" \
+        --image "$(sqlServerImage)" --colima > "$SQL_LOG" 2>&1 &
       SQL_PID=$!
 
       echo "Installing Python dependencies (overlapped with container setup)..."
@@ -591,12 +563,12 @@ jobs:
       ( cd mssql_python/pybind && ./build.sh )
 
       echo "Waiting for container setup (Colima + SQL Server) to finish..."
-      if ! wait "$SQL_PID"; then
-        echo "Container setup failed:"
-        cat /tmp/sql_setup.log
-        exit 1
+      SQL_STATUS=0
+      wait "$SQL_PID" || SQL_STATUS=$?
+      SQL_PID=
+      if [ "$SQL_STATUS" -ne 0 ]; then
+        exit "$SQL_STATUS"
       fi
-      cat /tmp/sql_setup.log
     displayName: 'Build + start SQL Server (Colima boot & SQL setup overlapped with build)'
     env:
       DB_PASSWORD: $(DB_PASSWORD)
@@ -729,6 +701,15 @@ jobs:
     condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/main'), or(eq(variables['sqlVersion'], 'SQL2022'), eq(variables['sqlVersion'], 'SQL2025')))
     continueOnError: true
 
+  - script: |
+      python3 eng/scripts/setup_sql_container.py --cleanup --colima \
+        --name sqlserver --owner "$(Build.BuildId).$(System.JobId)"
+    displayName: 'Clean up macOS SQL container'
+    condition: always()
+    timeoutInMinutes: 3
+    env:
+      DB_PASSWORD: $(DB_PASSWORD)
+
 - job: PytestOnLinux
   displayName: 'Linux x86_64'
   pool:
@@ -775,56 +756,12 @@ jobs:
     displayName: 'Create $(distroName) container'
 
   - script: |
-      # Start SQL Server container
-      docker run -d --name sqlserver-$(distroName) \
-        -e ACCEPT_EULA=Y \
-        -e MSSQL_SA_PASSWORD="$(DB_PASSWORD)" \
-        -p 1433:1433 \
-        $(sqlServerImage)
-      
-      # Wait for SQL Server to be ready
-      echo "Waiting for SQL Server to start..."
-      sql_ready=false
-      for i in {1..60}; do
-        if docker exec sqlserver-$(distroName) \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-          sql_ready=true
-          break
-        fi
-        echo "Waiting... ($i/60)"
-        sleep 2
-      done
-      if [ "$sql_ready" != true ]; then
-        # the loop sleeps after its final attempt, so probe once more
-        if docker exec sqlserver-$(distroName) \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-        else
-          echo "SQL Server did not become ready after 60 attempts"
-          echo "---- docker logs sqlserver-$(distroName) (last 200 lines) ----"
-          docker logs --tail 200 sqlserver-$(distroName) || true
-          exit 1
-        fi
-      fi
-      
-      # Create test database
-      docker exec sqlserver-$(distroName) \
-        /opt/mssql-tools18/bin/sqlcmd \
-        -S localhost \
-        -U SA \
-        -P "$(DB_PASSWORD)" \
-        -C -Q "CREATE DATABASE TestDB"
+      python3 eng/scripts/setup_sql_container.py \
+        --name "sqlserver-$(distroName)" --owner "$(Build.BuildId).$(System.JobId)" \
+        --image "$(sqlServerImage)" --database TestDB
     displayName: 'Start SQL Server container for $(distroName)'
-    condition: eq(variables['useAzureSQL'], 'false')
+    condition: and(succeeded(), eq(variables['useAzureSQL'], 'false'))
+    timeoutInMinutes: 22
     env:
       DB_PASSWORD: $(DB_PASSWORD)
 
@@ -1065,13 +1002,17 @@ jobs:
     condition: always()
 
   - script: |
-      # Clean up containers
+      python3 eng/scripts/setup_sql_container.py --cleanup \
+        --name "sqlserver-$(distroName)" --owner "$(Build.BuildId).$(System.JobId)"
+    displayName: 'Clean up sqlserver-$(distroName) SQL container'
+    condition: and(always(), eq(variables['useAzureSQL'], 'false'))
+    timeoutInMinutes: 3
+    env:
+      DB_PASSWORD: $(DB_PASSWORD)
+
+  - script: |
       docker stop test-container-$(distroName) || true
       docker rm test-container-$(distroName) || true
-      if [ "$(useAzureSQL)" = "false" ]; then
-        docker stop sqlserver-$(distroName) || true
-        docker rm sqlserver-$(distroName) || true
-      fi
     displayName: 'Clean up $(distroName) containers'
     condition: always()
 
@@ -1121,56 +1062,11 @@ jobs:
     displayName: 'Create $(distroName) ARM64 container'
 
   - script: |
-      # Start SQL Server container (x86_64 - SQL Server doesn't support ARM64)
-      docker run -d --name sqlserver-$(distroName)-$(archName) \
-        --platform linux/amd64 \
-        -e ACCEPT_EULA=Y \
-        -e MSSQL_SA_PASSWORD="$(DB_PASSWORD)" \
-        -p 1433:1433 \
-        mcr.microsoft.com/mssql/server:2022-latest
-      
-      # Wait for SQL Server to be ready
-      echo "Waiting for SQL Server to start..."
-      sql_ready=false
-      for i in {1..60}; do
-        if docker exec sqlserver-$(distroName)-$(archName) \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-          sql_ready=true
-          break
-        fi
-        echo "Waiting... ($i/60)"
-        sleep 2
-      done
-      if [ "$sql_ready" != true ]; then
-        # the loop sleeps after its final attempt, so probe once more
-        if docker exec sqlserver-$(distroName)-$(archName) \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-        else
-          echo "SQL Server did not become ready after 60 attempts"
-          echo "---- docker logs sqlserver-$(distroName)-$(archName) (last 200 lines) ----"
-          docker logs --tail 200 sqlserver-$(distroName)-$(archName) || true
-          exit 1
-        fi
-      fi
-      
-      # Create test database
-      docker exec sqlserver-$(distroName)-$(archName) \
-        /opt/mssql-tools18/bin/sqlcmd \
-        -S localhost \
-        -U SA \
-        -P "$(DB_PASSWORD)" \
-        -C -Q "CREATE DATABASE TestDB"
+      python3 eng/scripts/setup_sql_container.py \
+        --name "sqlserver-$(distroName)-$(archName)" --owner "$(Build.BuildId).$(System.JobId)" \
+        --image "mcr.microsoft.com/mssql/server:2022-latest" --database TestDB
     displayName: 'Start SQL Server container for $(distroName) ARM64'
+    timeoutInMinutes: 22
     env:
       DB_PASSWORD: $(DB_PASSWORD)
 
@@ -1339,11 +1235,17 @@ jobs:
     condition: always()
 
   - script: |
-      # Clean up containers
+      python3 eng/scripts/setup_sql_container.py --cleanup \
+        --name "sqlserver-$(distroName)-$(archName)" --owner "$(Build.BuildId).$(System.JobId)"
+    displayName: 'Clean up sqlserver-$(distroName)-$(archName) SQL container'
+    condition: always()
+    timeoutInMinutes: 3
+    env:
+      DB_PASSWORD: $(DB_PASSWORD)
+
+  - script: |
       docker stop test-container-$(distroName)-$(archName) || true
       docker rm test-container-$(distroName)-$(archName) || true
-      docker stop sqlserver-$(distroName)-$(archName) || true
-      docker rm sqlserver-$(distroName)-$(archName) || true
     displayName: 'Clean up $(distroName) ARM64 containers'
     condition: always()
 
@@ -1370,55 +1272,11 @@ jobs:
     displayName: 'Create RHEL 9 container'
 
   - script: |
-      # Start SQL Server container
-      docker run -d --name sqlserver-rhel9 \
-        -e ACCEPT_EULA=Y \
-        -e MSSQL_SA_PASSWORD="$(DB_PASSWORD)" \
-        -p 1433:1433 \
-        mcr.microsoft.com/mssql/server:2022-latest
-      
-      # Wait for SQL Server to be ready
-      echo "Waiting for SQL Server to start..."
-      sql_ready=false
-      for i in {1..60}; do
-        if docker exec sqlserver-rhel9 \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-          sql_ready=true
-          break
-        fi
-        echo "Waiting... ($i/60)"
-        sleep 2
-      done
-      if [ "$sql_ready" != true ]; then
-        # the loop sleeps after its final attempt, so probe once more
-        if docker exec sqlserver-rhel9 \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-        else
-          echo "SQL Server did not become ready after 60 attempts"
-          echo "---- docker logs sqlserver-rhel9 (last 200 lines) ----"
-          docker logs --tail 200 sqlserver-rhel9 || true
-          exit 1
-        fi
-      fi
-      
-      # Create test database
-      docker exec sqlserver-rhel9 \
-        /opt/mssql-tools18/bin/sqlcmd \
-        -S localhost \
-        -U SA \
-        -P "$(DB_PASSWORD)" \
-        -C -Q "CREATE DATABASE TestDB"
+      python3 eng/scripts/setup_sql_container.py \
+        --name "sqlserver-rhel9" --owner "$(Build.BuildId).$(System.JobId)" \
+        --image "mcr.microsoft.com/mssql/server:2022-latest" --database TestDB
     displayName: 'Start SQL Server container for RHEL 9'
+    timeoutInMinutes: 22
     env:
       DB_PASSWORD: $(DB_PASSWORD)
 
@@ -1575,11 +1433,17 @@ jobs:
     condition: always()
 
   - script: |
-      # Clean up containers
+      python3 eng/scripts/setup_sql_container.py --cleanup \
+        --name "sqlserver-rhel9" --owner "$(Build.BuildId).$(System.JobId)"
+    displayName: 'Clean up sqlserver-rhel9 SQL container'
+    condition: always()
+    timeoutInMinutes: 3
+    env:
+      DB_PASSWORD: $(DB_PASSWORD)
+
+  - script: |
       docker stop test-container-rhel9 || true
       docker rm test-container-rhel9 || true
-      docker stop sqlserver-rhel9 || true
-      docker rm sqlserver-rhel9 || true
     displayName: 'Clean up RHEL 9 containers'
     condition: always()
 
@@ -1614,56 +1478,11 @@ jobs:
     displayName: 'Create RHEL 9 ARM64 container'
 
   - script: |
-      # Start SQL Server container (x86_64 - SQL Server doesn't support ARM64)
-      docker run -d --name sqlserver-rhel9-arm64 \
-        --platform linux/amd64 \
-        -e ACCEPT_EULA=Y \
-        -e MSSQL_SA_PASSWORD="$(DB_PASSWORD)" \
-        -p 1433:1433 \
-        mcr.microsoft.com/mssql/server:2022-latest
-      
-      # Wait for SQL Server to be ready
-      echo "Waiting for SQL Server to start..."
-      sql_ready=false
-      for i in {1..60}; do
-        if docker exec sqlserver-rhel9-arm64 \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-          sql_ready=true
-          break
-        fi
-        echo "Waiting... ($i/60)"
-        sleep 2
-      done
-      if [ "$sql_ready" != true ]; then
-        # the loop sleeps after its final attempt, so probe once more
-        if docker exec sqlserver-rhel9-arm64 \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-        else
-          echo "SQL Server did not become ready after 60 attempts"
-          echo "---- docker logs sqlserver-rhel9-arm64 (last 200 lines) ----"
-          docker logs --tail 200 sqlserver-rhel9-arm64 || true
-          exit 1
-        fi
-      fi
-      
-      # Create test database
-      docker exec sqlserver-rhel9-arm64 \
-        /opt/mssql-tools18/bin/sqlcmd \
-        -S localhost \
-        -U SA \
-        -P "$(DB_PASSWORD)" \
-        -C -Q "CREATE DATABASE TestDB"
+      python3 eng/scripts/setup_sql_container.py \
+        --name "sqlserver-rhel9-arm64" --owner "$(Build.BuildId).$(System.JobId)" \
+        --image "mcr.microsoft.com/mssql/server:2022-latest" --database TestDB
     displayName: 'Start SQL Server container for RHEL 9 ARM64'
+    timeoutInMinutes: 22
     env:
       DB_PASSWORD: $(DB_PASSWORD)
 
@@ -1824,11 +1643,17 @@ jobs:
     condition: always()
 
   - script: |
-      # Clean up containers
+      python3 eng/scripts/setup_sql_container.py --cleanup \
+        --name "sqlserver-rhel9-arm64" --owner "$(Build.BuildId).$(System.JobId)"
+    displayName: 'Clean up sqlserver-rhel9-arm64 SQL container'
+    condition: always()
+    timeoutInMinutes: 3
+    env:
+      DB_PASSWORD: $(DB_PASSWORD)
+
+  - script: |
       docker stop test-container-rhel9-arm64 || true
       docker rm test-container-rhel9-arm64 || true
-      docker stop sqlserver-rhel9-arm64 || true
-      docker rm sqlserver-rhel9-arm64 || true
     displayName: 'Clean up RHEL 9 ARM64 containers'
     condition: always()
 
@@ -1863,56 +1688,11 @@ jobs:
     displayName: 'Create Alpine x86_64 container'
 
   - script: |
-      # Start SQL Server container (x86_64)
-      docker run -d --name sqlserver-alpine \
-        --platform linux/amd64 \
-        -e ACCEPT_EULA=Y \
-        -e MSSQL_SA_PASSWORD="$(DB_PASSWORD)" \
-        -p 1433:1433 \
-        mcr.microsoft.com/mssql/server:2022-latest
-      
-      # Wait for SQL Server to be ready
-      echo "Waiting for SQL Server to start..."
-      sql_ready=false
-      for i in {1..60}; do
-        if docker exec sqlserver-alpine \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-          sql_ready=true
-          break
-        fi
-        echo "Waiting... ($i/60)"
-        sleep 2
-      done
-      if [ "$sql_ready" != true ]; then
-        # the loop sleeps after its final attempt, so probe once more
-        if docker exec sqlserver-alpine \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-        else
-          echo "SQL Server did not become ready after 60 attempts"
-          echo "---- docker logs sqlserver-alpine (last 200 lines) ----"
-          docker logs --tail 200 sqlserver-alpine || true
-          exit 1
-        fi
-      fi
-      
-      # Create test database
-      docker exec sqlserver-alpine \
-        /opt/mssql-tools18/bin/sqlcmd \
-        -S localhost \
-        -U SA \
-        -P "$(DB_PASSWORD)" \
-        -C -Q "CREATE DATABASE TestDB"
+      python3 eng/scripts/setup_sql_container.py \
+        --name "sqlserver-alpine" --owner "$(Build.BuildId).$(System.JobId)" \
+        --image "mcr.microsoft.com/mssql/server:2022-latest" --database TestDB
     displayName: 'Start SQL Server container for Alpine x86_64'
+    timeoutInMinutes: 22
     env:
       DB_PASSWORD: $(DB_PASSWORD)
 
@@ -2096,11 +1876,17 @@ jobs:
     condition: always()
 
   - script: |
-      # Clean up containers
+      python3 eng/scripts/setup_sql_container.py --cleanup \
+        --name "sqlserver-alpine" --owner "$(Build.BuildId).$(System.JobId)"
+    displayName: 'Clean up sqlserver-alpine SQL container'
+    condition: always()
+    timeoutInMinutes: 3
+    env:
+      DB_PASSWORD: $(DB_PASSWORD)
+
+  - script: |
       docker stop test-container-alpine || true
       docker rm test-container-alpine || true
-      docker stop sqlserver-alpine || true
-      docker rm sqlserver-alpine || true
     displayName: 'Clean up Alpine x86_64 containers'
     condition: always()
 
@@ -2137,56 +1923,11 @@ jobs:
     displayName: 'Create Alpine ARM64 container'
 
   - script: |
-      # Start SQL Server container (x86_64 - SQL Server doesn't support ARM64)
-      docker run -d --name sqlserver-alpine-arm64 \
-        --platform linux/amd64 \
-        -e ACCEPT_EULA=Y \
-        -e MSSQL_SA_PASSWORD="$(DB_PASSWORD)" \
-        -p 1433:1433 \
-        mcr.microsoft.com/mssql/server:2022-latest
-      
-      # Wait for SQL Server to be ready
-      echo "Waiting for SQL Server to start..."
-      sql_ready=false
-      for i in {1..60}; do
-        if docker exec sqlserver-alpine-arm64 \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-          sql_ready=true
-          break
-        fi
-        echo "Waiting... ($i/60)"
-        sleep 2
-      done
-      if [ "$sql_ready" != true ]; then
-        # the loop sleeps after its final attempt, so probe once more
-        if docker exec sqlserver-alpine-arm64 \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1" >/dev/null 2>&1; then
-          echo "SQL Server is ready!"
-        else
-          echo "SQL Server did not become ready after 60 attempts"
-          echo "---- docker logs sqlserver-alpine-arm64 (last 200 lines) ----"
-          docker logs --tail 200 sqlserver-alpine-arm64 || true
-          exit 1
-        fi
-      fi
-      
-      # Create test database
-      docker exec sqlserver-alpine-arm64 \
-        /opt/mssql-tools18/bin/sqlcmd \
-        -S localhost \
-        -U SA \
-        -P "$(DB_PASSWORD)" \
-        -C -Q "CREATE DATABASE TestDB"
+      python3 eng/scripts/setup_sql_container.py \
+        --name "sqlserver-alpine-arm64" --owner "$(Build.BuildId).$(System.JobId)" \
+        --image "mcr.microsoft.com/mssql/server:2022-latest" --database TestDB
     displayName: 'Start SQL Server container for Alpine ARM64'
+    timeoutInMinutes: 22
     env:
       DB_PASSWORD: $(DB_PASSWORD)
 
@@ -2371,11 +2112,17 @@ jobs:
     condition: always()
 
   - script: |
-      # Clean up containers
+      python3 eng/scripts/setup_sql_container.py --cleanup \
+        --name "sqlserver-alpine-arm64" --owner "$(Build.BuildId).$(System.JobId)"
+    displayName: 'Clean up sqlserver-alpine-arm64 SQL container'
+    condition: always()
+    timeoutInMinutes: 3
+    env:
+      DB_PASSWORD: $(DB_PASSWORD)
+
+  - script: |
       docker stop test-container-alpine-arm64 || true
       docker rm test-container-alpine-arm64 || true
-      docker stop sqlserver-alpine-arm64 || true
-      docker rm sqlserver-alpine-arm64 || true
     displayName: 'Clean up Alpine ARM64 containers'
     condition: always()
 
@@ -2624,44 +2371,11 @@ jobs:
     displayName: 'Install build dependencies'
 
   - script: |
-      # Start SQL Server container
-      docker pull mcr.microsoft.com/mssql/server:2022-latest
-      docker run \
-        --name sqlserver \
-        -e ACCEPT_EULA=Y \
-        -e MSSQL_SA_PASSWORD="$(DB_PASSWORD)" \
-        -p 1433:1433 \
-        -d mcr.microsoft.com/mssql/server:2022-latest
-
-      # Wait until SQL Server is ready
-      sql_ready=false
-      for i in {1..30}; do
-        if docker exec sqlserver \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1"; then
-          sql_ready=true
-          break
-        fi
-        sleep 2
-      done
-      if [ "$sql_ready" != true ]; then
-        # the loop sleeps after its final attempt, so probe once more
-        if ! docker exec sqlserver \
-          /opt/mssql-tools18/bin/sqlcmd \
-          -S localhost \
-          -U SA \
-          -P "$(DB_PASSWORD)" \
-          -C -Q "SELECT 1"; then
-          echo "SQL Server did not become ready after 30 attempts"
-          echo "---- docker logs sqlserver (last 200 lines) ----"
-          docker logs --tail 200 sqlserver || true
-          exit 1
-        fi
-      fi
+      python3 eng/scripts/setup_sql_container.py \
+        --name "sqlserver" --owner "$(Build.BuildId).$(System.JobId)" \
+        --image "mcr.microsoft.com/mssql/server:2022-latest"
     displayName: 'Start SQL Server container'
+    timeoutInMinutes: 22
     env:
       DB_PASSWORD: $(DB_PASSWORD)
 
@@ -2708,3 +2422,12 @@ jobs:
       reportDirectory: 'unified-coverage'
       failIfCoverageEmpty: true
     displayName: 'Publish unified code coverage results'
+
+  - script: |
+      python3 eng/scripts/setup_sql_container.py --cleanup \
+        --name sqlserver --owner "$(Build.BuildId).$(System.JobId)"
+    displayName: 'Clean up coverage SQL container'
+    condition: always()
+    timeoutInMinutes: 3
+    env:
+      DB_PASSWORD: $(DB_PASSWORD)
diff --git a/eng/scripts/setup_sql_container.py b/eng/scripts/setup_sql_container.py
new file mode 100644
index 00000000..7beda1af
--- /dev/null
+++ b/eng/scripts/setup_sql_container.py
@@ -0,0 +1,826 @@
+#!/usr/bin/env python3
+"""Create a job-owned SQL Server container, retrying SQL setup once.
+
+Invoke on the Docker host with --name, --image and --owner; DB_PASSWORD is
+required for setup and is never an argument. Use --database TestDB on Linux
+test legs, --colima on macOS, and --cleanup for the always-running final step.
+Cleanup refuses containers without the matching owner label.
+
+Only SQL lookup/pull/create/start/readiness/database setup is retried. Colima
+starts once within the remaining overall setup budget, without a separate
+startup timeout. The owned-container lookup also checks Docker availability, so
+transient read-only failures can consume the same two attempts without creating
+or removing anything. Missing tools, invalid configuration/lookup results,
+ownership conflicts and unsafe cleanup are terminal.
+Linux/macOS attempts are bounded at 600/900 seconds including diagnostics and
+cleanup; entire invocations at 1260/2460 seconds including macOS VM startup.
+Command budgets include termination/output-drain grace. Cleanup-only uses a
+115-second deadline. OS scheduling/host loss can defeat cooperative deadlines;
+the pipeline also enforces outer task limits.
+
+Readiness polls for 120/180 seconds (Linux/macOS), then performs one final
+state/query check bounded at 45 seconds, for at most 165/225 seconds total.
+Both are clamped to the remaining attempt/global budget, preserving cleanup.
+
+Docker diagnostics request only the last 30 minutes and at most 5000 lines,
+within 20 seconds. Redacted output retains 32768-character head/tail excerpts
+plus at most 8192 characters of fatal/Reason context, not complete history.
+No dumps or full inspect output are collected. Timeout/cancellation targets only
+each command's original group and still-owned unreaped child, with at most four
+seconds of teardown within the original command budget. Hard host loss can
+still prevent cleanup; the pipeline also runs --cleanup.
+Permission or identity failures are reported as incomplete, nonretryable
+teardown without replacing the original failure or cancellation. Group checks
+are not atomic with signalling: departed groups are not followed, and a reaped
+PID observed again is not targeted. A missing leader alone does not establish
+that its original group is gone.
+
+Colima is a daemon launcher, not a finite-output command. Its direct exit is
+captured through a private anonymous temporary file, without requiring EOF or
+stopping successful background processes. Only a fixed-size startup snapshot
+is read/redacted; Docker/SQL readiness is checked separately. The parent closes
+its descriptor on every outcome. Daemons can retain the unlinked backing inode
+until their descriptors close or the hosted job ends: raw backing storage is
+not hard-capped. This contract is intended for job-scoped hosted Colima startup,
+not a general-purpose persistent daemon logging service.
+"""
+
+import argparse
+import codecs
+from dataclasses import dataclass
+import io
+import json
+import os
+import re
+import signal
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+
+OWNER_LABEL = "com.microsoft.mssql-python.ci-owner"
+STATE_FORMAT = (
+    '{{.Id}}|{{index .Config.Labels "' + OWNER_LABEL + '"}}|'
+    "{{.State.Status}}|{{.State.ExitCode}}|{{.State.OOMKilled}}|{{.Image}}"
+)
+SQLCMD = "/opt/mssql-tools18/bin/sqlcmd"
+
+
+class SetupFailure(Exception):
+    def __init__(self, message, *, retryable=True):
+        super().__init__(message)
+        self.retryable = retryable
+
+
+class SetupTimeout(SetupFailure):
+    pass
+
+
+class Cancelled(BaseException):
+    def __init__(self, signum):
+        self.signum = signum
+
+
+def redact(text, password):
+    if password:
+        text = text.replace(password, "[REDACTED]")
+    return re.sub(
+        r"(?i)\b(?:DB_PASSWORD|MSSQL_SA_PASSWORD|SQLCMDPASSWORD|PASSWORD|PWD)"
+        r"\s*[:=]\s*(?:\"[^\"]*\"|'[^']*'|[^;\r\n]*)",
+        "[credential redacted]",
+        text,
+    )
+
+
+class SafeCapture:
+    """Redact complete lines before retaining bounded head/tail excerpts."""
+
+    def __init__(self, password, limit=32768):
+        self.password = password
+        self.limit = limit
+        self.head = ""
+        self.tail = ""
+        self.total = 0
+        self.fatal_context = ""
+        self.context_remaining = 0
+
+    def add(self, line):
+        safe = redact(line, self.password)
+        if re.search(r"\bfatal\b|\bReason:", safe, re.IGNORECASE):
+            self.context_remaining = 12
+        if self.context_remaining:
+            remaining = min(8192, self.limit // 4) - len(self.fatal_context)
+            self.fatal_context += safe[:remaining]
+            self.context_remaining -= 1
+        self.total += len(safe)
+        remaining = self.limit - len(self.head)
+        self.head += safe[:remaining]
+        self.tail = (self.tail + safe[remaining:])[-self.limit :]
+
+    def read(self, stream):
+        decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
+        pending = ""
+        omitted = False
+        read = getattr(stream, "read1", stream.read)
+        while True:
+            chunk = read(4096)
+            text = decoder.decode(chunk, final=not chunk)
+            for part in text.splitlines(keepends=True):
+                pending += part
+                if len(pending) > 8192:
+                    if not omitted:
+                        self.add("[overlong diagnostic line omitted]\n")
+                    pending = ""
+                    omitted = True
+                if part.endswith(("\n", "\r")):
+                    if not omitted:
+                        self.add(pending)
+                    pending = ""
+                    omitted = False
+            if not chunk:
+                if pending and not omitted:
+                    self.add(pending)
+                return
+
+    def output(self):
+        marker = "\n[diagnostic output truncated]\n" if self.total > 2 * self.limit else ""
+        context = (
+            "\n[fatal/Reason context from retrieved output]\n" + self.fatal_context
+            if marker and self.fatal_context
+            else ""
+        )
+        return self.head + marker + context + self.tail
+
+
+@dataclass
+class Result:
+    returncode: int
+    output: str
+
+
+@dataclass
+class StopResult:
+    reaped: bool
+    errors: list[str]
+
+
+class Commands:
+    def __init__(self, password):
+        self.password = password
+
+    @staticmethod
+    def stop(process, deadline):
+        errors = []
+        group_blocked = False
+
+        def signal_child(name):
+            # Popen rechecks its own unreaped child before signalling its PID.
+            if process.poll() is not None:
+                return
+            try:
+                if name == "TERM":
+                    process.terminate()
+                else:
+                    process.kill()
+            except ProcessLookupError:
+                pass
+            except PermissionError as exc:
+                errors.append(
+                    f"Permission denied sending SIG{name} to owned child "
+                    f"{process.pid} (errno={exc.errno})"
+                )
+
+        def signal_owned(name):
+            nonlocal group_blocked
+            if os.name != "posix":
+                signal_child(name)
+                return
+            if group_blocked:
+                signal_child(name)
+                return
+            reaped = process.poll() is not None
+            try:
+                group = os.getpgid(process.pid)
+            except ProcessLookupError:
+                # Descendants may still hold the original group and stdout.
+                pass
+            except PermissionError as exc:
+                errors.append(
+                    f"Permission denied checking original group {process.pid} (errno={exc.errno})"
+                )
+                group_blocked = True
+            else:
+                if reaped:
+                    errors.append(f"Reaped PID {process.pid} is present again; group not signalled")
+                    group_blocked = True
+                elif group != process.pid:
+                    errors.append(
+                        f"Owned child {process.pid} left its original group; "
+                        f"group {group} not followed"
+                    )
+                    group_blocked = True
+            if group_blocked:
+                signal_child(name)
+                return
+            try:
+                os.killpg(process.pid, getattr(signal, "SIG" + name))
+            except ProcessLookupError:
+                signal_child(name)
+            except PermissionError as exc:
+                errors.append(
+                    f"Permission denied sending SIG{name} to original group "
+                    f"{process.pid} (errno={exc.errno})"
+                )
+                group_blocked = True
+                signal_child(name)
+
+        signal_owned("TERM")
+        try:
+            process.wait(timeout=max(0, min(1, (deadline - time.monotonic()) / 2)))
+        except subprocess.TimeoutExpired:
+            pass
+        signal_owned("KILL")
+        try:
+            process.wait(timeout=max(0, min(1, deadline - time.monotonic())))
+        except subprocess.TimeoutExpired:
+            return StopResult(False, errors)
+        return StopResult(True, errors)
+
+    @staticmethod
+    def snapshot(output, capture):
+        output.flush()
+        size = os.fstat(output.fileno()).st_size
+        window = 65536
+        ranges = [(0, size)] if size <= 2 * window else [(0, window), (size - window, window)]
+        capture.add(
+            "[Colima startup snapshot: at most 64KiB head and 64KiB tail; "
+            "partial boundary lines and later background output omitted]\n"
+        )
+        for index, (offset, length) in enumerate(ranges):
+            if index:
+                capture.context_remaining = 0
+                capture.add("[Colima startup snapshot truncated; middle output omitted]\n")
+            if hasattr(os, "pread"):
+                data = os.pread(output.fileno(), length, offset)
+            else:
+                output.seek(offset)
+                data = output.read(length)
+            if offset:
+                data = data.partition(b"\n")[2]
+            if data and not data.endswith(b"\n"):
+                data = data[: data.rfind(b"\n") + 1]
+            capture.read(io.BytesIO(data))
+
+    def run_launcher(self, args, timeout, *, env=None):
+        deadline = time.monotonic() + timeout
+        try:
+            output = tempfile.TemporaryFile(mode="a+b")
+        except OSError as exc:
+            raise SetupFailure(
+                f"Cannot create private Colima output capture (errno={exc.errno})", retryable=False
+            ) from None
+        primary = None
+        try:
+            return self._run(args, deadline - time.monotonic(), env=env, output_file=output)
+        except (SetupFailure, Cancelled) as exc:
+            primary = exc
+            raise
+        finally:
+            try:
+                output.close()
+            except OSError as exc:
+                message = f"Cannot close private Colima output capture (errno={exc.errno})"
+                if primary is None:
+                    raise SetupFailure(message, retryable=False) from None
+                if isinstance(primary, SetupFailure):
+                    primary.retryable = False
+                    primary.args = (str(primary) + "\n" + message,)
+                else:
+                    print("[sql] " + message, file=sys.stderr, flush=True)
+
+    def run(self, args, timeout, *, env=None):
+        return self._run(args, timeout, env=env)
+
+    def _run(self, args, timeout, *, env=None, output_file=None):
+        if timeout <= 0:
+            raise SetupTimeout("SQL setup deadline exhausted")
+        deadline = time.monotonic() + timeout
+        grace = min(4, timeout / 2)
+        work_deadline = deadline - grace
+        capture = SafeCapture(self.password)
+        try:
+            process = subprocess.Popen(
+                args,
+                stdout=subprocess.PIPE if output_file is None else output_file,
+                stderr=subprocess.STDOUT,
+                env=env,
+                start_new_session=os.name == "posix",
+            )
+        except OSError:
+            raise SetupFailure("Cannot launch required setup command", retryable=False) from None
+        output_done = threading.Event()
+        read_errors = ["Setup output reader did not complete"] if output_file is None else []
+
+        def read_output():
+            try:
+                capture.read(process.stdout)
+            except OSError:
+                read_errors[:] = ["Cannot read setup command output"]
+            else:
+                read_errors.clear()
+            finally:
+                output_done.set()
+
+        reader = threading.Thread(target=read_output, daemon=True) if output_file is None else None
+        reader_started = False
+        timed_out = False
+        drain_timed_out = False
+        stopped = StopResult(True, [])
+        pending_error = None
+        problems = []
+
+        def failure_message(reason):
+            message = reason
+            if problems:
+                message += "\nCommand teardown incomplete:\n" + "\n".join(problems)
+            message += "\nCaptured command output (may be incomplete):"
+            message += "\n" + (capture.output() or "[no completed output lines captured]")
+            return redact(message, self.password)
+
+        try:
+            if reader is not None:
+                try:
+                    reader.start()
+                    reader_started = True
+                except RuntimeError:
+                    raise SetupFailure(
+                        "Cannot start setup output reader", retryable=False
+                    ) from None
+            try:
+                process.wait(timeout=max(0, work_deadline - time.monotonic()))
+            except subprocess.TimeoutExpired:
+                timed_out = True
+            if not timed_out and output_file is None:
+                drain_timed_out = not output_done.wait(
+                    timeout=max(0, work_deadline - time.monotonic())
+                )
+        except (Cancelled, SetupFailure) as exc:
+            pending_error = exc
+            raise
+        finally:
+            teardown_deadline = min(deadline, time.monotonic() + 4)
+            if (
+                process.poll() is None
+                or (output_file is None and (not output_done.is_set() or read_errors))
+                or (
+                    output_file is not None
+                    and (pending_error is not None or process.returncode != 0)
+                )
+            ):
+                stopped = self.stop(process, teardown_deadline)
+            problems.extend(stopped.errors)
+            if reader_started:
+                output_done.wait(timeout=max(0, teardown_deadline - time.monotonic()))
+            if output_file is not None:
+                try:
+                    self.snapshot(output_file, capture)
+                except OSError as exc:
+                    problems.append(f"Cannot read Colima startup snapshot (errno={exc.errno})")
+                output_done.set()
+            elif not reader_started or output_done.is_set():
+                try:
+                    process.stdout.close()
+                except OSError as exc:
+                    problems.append(f"Cannot close command output (errno={exc.errno})")
+            if not stopped.reaped:
+                problems.append("Owned child could not be reaped within the teardown deadline")
+            if reader_started and not output_done.is_set():
+                problems.append("Output reader did not finish within the teardown deadline")
+            elif reader_started and read_errors:
+                problems.append("Output capture error: " + read_errors[0])
+            if pending_error is not None:
+                reason = (
+                    f"Command cancelled (signal {pending_error.signum})"
+                    if isinstance(pending_error, Cancelled)
+                    else str(pending_error)
+                )
+                print("[sql] " + failure_message(reason), file=sys.stderr, flush=True)
+        if timed_out:
+            raise SetupTimeout(
+                failure_message("Setup command timed out"),
+                retryable=not problems and not read_errors,
+            )
+        if drain_timed_out or not output_done.is_set():
+            raise SetupFailure(
+                failure_message("Setup command output drain timed out"), retryable=False
+            )
+        if read_errors:
+            raise SetupFailure(failure_message(read_errors[0]), retryable=False)
+        if problems:
+            reason = (
+                f"Setup command failed (exit {process.returncode})"
+                if process.returncode
+                else "Setup command teardown failed"
+            )
+            raise SetupFailure(failure_message(reason), retryable=False)
+        return Result(process.returncode, capture.output())
+
+
+@dataclass
+class Container:
+    identifier: str
+    status: str
+    exit_code: str
+    oom_killed: str
+    image: str
+
+
+class SqlSetup:
+    def __init__(self, args, password, commands=None):
+        self.args = args
+        self.password = password
+        self.commands = commands or Commands(password)
+        self.deadline = time.monotonic() + (115 if args.cleanup else 2460 if args.colima else 1260)
+        self.phase_deadline = self.deadline
+        self.container = None
+        self.creation_requested = False
+        self.image_id = None
+        self.docker = ["docker"] + (["--context", "colima"] if args.colima else [])
+        self.env = os.environ.copy()
+        for key in ("DB_PASSWORD", "DB_CONNECTION_STRING", "MSSQL_SA_PASSWORD", "SQLCMDPASSWORD"):
+            self.env.pop(key, None)
+        self.env.update(MSSQL_SA_PASSWORD=password, SQLCMDPASSWORD=password)
+
+    def log(self, message):
+        print("[sql] " + redact(message, self.password), flush=True)
+
+    def command(self, args, timeout=15, *, check=True, deadline=None, launcher=False):
+        end = min(self.deadline, self.phase_deadline if deadline is None else deadline)
+        remaining = min(timeout, end - time.monotonic())
+        if remaining <= 0:
+            raise SetupTimeout("SQL setup deadline exhausted")
+        run = self.commands.run_launcher if launcher else self.commands.run
+        result = run(args, remaining, env=self.env)
+        if result.returncode in (-signal.SIGINT, -signal.SIGTERM, 130, 143):
+            raise Cancelled(
+                signal.SIGINT if result.returncode in (-signal.SIGINT, 130) else signal.SIGTERM
+            )
+        if check and result.returncode != 0:
+            raise SetupFailure(f"Setup command failed (exit {result.returncode})\n{result.output}")
+        return result
+
+    def docker_command(self, *args, **kwargs):
+        return self.command(self.docker + list(args), **kwargs)
+
+    def find_owned(self, *, deadline=None):
+        result = self.docker_command(
+            "container",
+            "ls",
+            "--all",
+            "--no-trunc",
+            "--filter",
+            f"name=^/{re.escape(self.args.name)}$",
+            "--format",
+            "{{.ID}}",
+            deadline=deadline,
+        )
+        identifiers = result.output.split()
+        if not identifiers:
+            return None
+        if len(identifiers) != 1 or not re.fullmatch(r"[0-9a-f]{64}", identifiers[0]):
+            raise SetupFailure("Unexpected exact-name container lookup result", retryable=False)
+        result = self.docker_command(
+            "inspect", "--format", STATE_FORMAT, identifiers[0], deadline=deadline
+        )
+        fields = result.output.strip().split("|")
+        if len(fields) != 6 or fields[0] != identifiers[0]:
+            raise SetupFailure("Invalid selected container state", retryable=False)
+        if fields[1] != self.args.owner:
+            raise SetupFailure(
+                "Container name is owned by another job or unlabelled", retryable=False
+            )
+        return Container(fields[0], fields[2], fields[3], fields[4], fields[5])
+
+    def diagnostics(self, container, *, deadline):
+        self.log(
+            f"Container {container.identifier}: status={container.status} "
+            f"exit={container.exit_code} OOMKilled={container.oom_killed} image={container.image}"
+        )
+        self.log(
+            "Container log excerpts (last 30m, max 5000 lines; may omit older history; "
+            "retained output may be truncated):"
+        )
+        try:
+            result = self.docker_command(
+                "logs",
+                "--since",
+                "30m",
+                "--tail",
+                "5000",
+                container.identifier,
+                timeout=20,
+                check=False,
+                deadline=deadline,
+            )
+            self.log(result.output)
+            if result.returncode:
+                self.log(f"Container logs unavailable (exit {result.returncode})")
+        except SetupFailure as exc:
+            self.log(f"Container logs unavailable: {exc}")
+            if not exc.retryable:
+                raise
+
+    def remove(self, container, *, deadline):
+        self.docker_command("rm", "--force", container.identifier, timeout=30, deadline=deadline)
+        remaining = self.find_owned(deadline=deadline)
+        if remaining is not None:
+            raise SetupFailure("Owned container still exists after removal", retryable=False)
+        self.container = None
+
+    def remove_with_evidence(self, container, *, deadline, diagnostic_deadline):
+        diagnostic_error = None
+        try:
+            self.diagnostics(container, deadline=diagnostic_deadline)
+        except (SetupFailure, Cancelled) as exc:
+            diagnostic_error = exc
+        try:
+            self.remove(container, deadline=deadline)
+        except (SetupFailure, Cancelled) as removal_error:
+            if diagnostic_error is None:
+                raise
+            if isinstance(diagnostic_error, Cancelled):
+                self.log(
+                    "Owned removal also failed during diagnostic cancellation: "
+                    + (
+                        f"signal {removal_error.signum}"
+                        if isinstance(removal_error, Cancelled)
+                        else str(removal_error)
+                    )
+                )
+                raise diagnostic_error from None
+            if isinstance(removal_error, Cancelled):
+                self.log(f"Diagnostics also failed before removal cancellation: {diagnostic_error}")
+                raise
+            raise SetupFailure(
+                f"Diagnostics failed: {diagnostic_error}\nOwned removal failed: {removal_error}",
+                retryable=False,
+            ) from None
+        if diagnostic_error is not None:
+            raise diagnostic_error
+
+    def cleanup(self, *, evidence=True):
+        # Include lookup/ownership checks and removal verification in addition
+        # to the log and rm deadlines.
+        self.phase_deadline = min(self.deadline, time.monotonic() + 100)
+        container = self.find_owned()
+        if container is None:
+            self.container = None
+            return
+        if evidence:
+            self.remove_with_evidence(
+                container,
+                deadline=self.phase_deadline,
+                diagnostic_deadline=min(self.phase_deadline - 30, time.monotonic() + 20),
+            )
+        else:
+            self.remove(container, deadline=self.phase_deadline)
+
+    def prepare_runtime(self):
+        if self.args.colima and not self.args.cleanup:
+            self.log("Starting Colima once (outside SQL retry)")
+            result = self.command(
+                ["colima", "start", "--cpu", "4", "--memory", "8", "--disk", "50"],
+                self.deadline - time.monotonic(),
+                launcher=True,
+            )
+            self.log("Colima launcher completed; Docker and SQL readiness still require checks")
+            self.log(result.output)
+
+    def acquire_image(self):
+        if self.image_id is not None:
+            return
+        self.docker_command(
+            "pull",
+            "--quiet",
+            "--platform",
+            "linux/amd64",
+            self.args.image,
+            timeout=600 if self.args.colima else 300,
+        )
+        result = self.docker_command(
+            "image", "inspect", "--format", "{{.Id}} {{json .RepoDigests}}", self.args.image
+        )
+        parts = result.output.strip().split(" ", 1)
+        if len(parts) != 2 or not re.fullmatch(r"sha256:[0-9a-f]{64}", parts[0]):
+            raise SetupFailure("Invalid resolved SQL image", retryable=False)
+        try:
+            digests = json.loads(parts[1])
+        except json.JSONDecodeError:
+            raise SetupFailure("Invalid SQL image digest metadata", retryable=False) from None
+        if not isinstance(digests, list) or not all(isinstance(item, str) for item in digests):
+            raise SetupFailure("Missing SQL image digest metadata", retryable=False)
+        self.image_id = parts[0]
+        self.log(f"SQL image {self.image_id}; repository digests={json.dumps(digests)}")
+
+    def sql(self, query, *, timeout=15, query_timeout=5, deadline=None):
+        return self.docker_command(
+            "exec",
+            "--env",
+            "SQLCMDPASSWORD",
+            self.container.identifier,
+            SQLCMD,
+            "-S",
+            "localhost",
+            "-U",
+            "SA",
+            "-C",
+            "-b",
+            "-l",
+            "5",
+            "-t",
+            str(query_timeout),
+            "-Q",
+            query,
+            timeout=timeout,
+            check=False,
+            deadline=deadline,
+        )
+
+    def readiness_probe(self, deadline):
+        current = self.find_owned(deadline=deadline)
+        if current is None or current.identifier != self.container.identifier:
+            raise SetupFailure("SQL container disappeared or was replaced", retryable=False)
+        if current.status != "running":
+            raise SetupFailure(
+                f"SQL container exited before readiness (status={current.status}, "
+                f"exit={current.exit_code}, OOMKilled={current.oom_killed})"
+            )
+        probe = self.sql("SELECT 1", deadline=deadline)
+        if probe.returncode in (126, 127):
+            raise SetupFailure("Required sqlcmd executable is unavailable", retryable=False)
+        return probe
+
+    def wait_ready(self):
+        polling_deadline = min(
+            self.deadline,
+            self.phase_deadline,
+            time.monotonic() + (180 if self.args.colima else 120),
+        )
+        while time.monotonic() < polling_deadline:
+            try:
+                probe = self.readiness_probe(polling_deadline)
+            except SetupTimeout as exc:
+                if not exc.retryable:
+                    raise
+                self.log(f"Readiness probe timed out: {exc}")
+            else:
+                if probe.returncode == 0:
+                    return
+            time.sleep(max(0, min(2, polling_deadline - time.monotonic())))
+        final_deadline = min(self.deadline, self.phase_deadline, polling_deadline + 45)
+        if time.monotonic() >= final_deadline:
+            raise SetupTimeout("SQL readiness budget exhausted before final probe")
+        self.log("Polling window ended; performing one final bounded SQL readiness check")
+        probe = self.readiness_probe(final_deadline)
+        if probe.returncode != 0:
+            raise SetupFailure("SQL final readiness check failed\n" + probe.output)
+
+    def attempt(self):
+        stale = self.find_owned()
+        if stale is not None:
+            self.log("Removing pre-existing same-job container before fresh setup")
+            try:
+                self.remove_with_evidence(
+                    stale,
+                    deadline=self.phase_deadline,
+                    diagnostic_deadline=min(self.phase_deadline, time.monotonic() + 20),
+                )
+            except SetupFailure as exc:
+                raise SetupFailure(str(exc), retryable=False) from None
+        self.acquire_image()
+        # The daemon may create the container even when the CLI times out.
+        self.creation_requested = True
+        self.docker_command(
+            "create",
+            "--name",
+            self.args.name,
+            "--label",
+            f"{OWNER_LABEL}={self.args.owner}",
+            "--platform",
+            "linux/amd64",
+            "--env",
+            "ACCEPT_EULA=Y",
+            "--env",
+            "MSSQL_SA_PASSWORD",
+            "-p",
+            "1433:1433",
+            self.image_id,
+            timeout=30,
+        )
+        self.container = self.find_owned()
+        if self.container is None:
+            raise SetupFailure("Created SQL container was not found")
+        self.docker_command("start", self.container.identifier, timeout=30)
+        self.wait_ready()
+        if self.args.database:
+            result = self.sql("CREATE DATABASE TestDB", timeout=30, query_timeout=15)
+            if result.returncode != 0:
+                raise SetupFailure("TestDB initialization failed\n" + result.output)
+
+    def setup(self):
+        self.prepare_runtime()
+        if self.args.cleanup:
+            self.cleanup(evidence=False)
+            self.log("Owned SQL container cleanup complete (or already absent)")
+            return
+        for number in (1, 2):
+            self.creation_requested = False
+            self.log(f"SQL setup attempt {number}/2")
+            attempt_end = min(
+                self.deadline,
+                time.monotonic() + (900 if self.args.colima else 600),
+            )
+            self.phase_deadline = attempt_end - 100
+            try:
+                self.attempt()
+            except SetupFailure as exc:
+                self.log(f"Attempt {number}/2 failed: {exc}")
+                self.phase_deadline = attempt_end
+                if self.creation_requested:
+                    try:
+                        self.cleanup()
+                    except SetupFailure as cleanup_error:
+                        raise SetupFailure(
+                            f"SQL setup failed: {exc}\nCannot safely recover/clean up: {cleanup_error}",
+                            retryable=False,
+                        ) from None
+                if not exc.retryable or number == 2:
+                    raise SetupFailure("SQL setup failed; no further attempts", retryable=False)
+                self.phase_deadline = self.deadline
+                if self.deadline - time.monotonic() < 5:
+                    raise SetupFailure("SQL setup deadline exhausted before retry", retryable=False)
+                self.log("Retrying SQL setup only after 5 seconds")
+                time.sleep(5)
+            else:
+                outcome = "ready on first attempt" if number == 1 else "recovered on second attempt"
+                self.log(f"SQL Server {outcome}")
+                return
+
+
+def arguments(argv):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--name", required=True)
+    parser.add_argument("--owner", required=True)
+    parser.add_argument("--image")
+    parser.add_argument("--database", choices=("TestDB",))
+    parser.add_argument("--colima", action="store_true")
+    parser.add_argument("--cleanup", action="store_true")
+    args = parser.parse_args(argv)
+    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]*", args.name):
+        raise SetupFailure("Invalid SQL container name", retryable=False)
+    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]*", args.owner):
+        raise SetupFailure("Invalid SQL container owner", retryable=False)
+    if not args.cleanup and (
+        not args.image
+        or not re.fullmatch(r"mcr\.microsoft\.com/mssql/server:[A-Za-z0-9_.-]+", args.image)
+    ):
+        raise SetupFailure(
+            "An explicit supported SQL Server image tag is required", retryable=False
+        )
+    return args
+
+
+def main(argv=None):
+    password = os.environ.get("DB_PASSWORD", "")
+    setup = None
+
+    def cancel(signum, _frame):
+        raise Cancelled(signum)
+
+    for signum in (signal.SIGINT, signal.SIGTERM):
+        signal.signal(signum, cancel)
+    try:
+        args = arguments(argv)
+        if not args.cleanup and (not password or "\n" in password or "\r" in password):
+            raise SetupFailure("DB_PASSWORD must be set to a single-line value", retryable=False)
+        setup = SqlSetup(args, password)
+        setup.setup()
+        return 0
+    except Cancelled as exc:
+        for signum in (signal.SIGINT, signal.SIGTERM):
+            signal.signal(signum, signal.SIG_IGN)
+        print("[sql] Setup cancelled; no retry", flush=True)
+        if setup is not None:
+            try:
+                setup.cleanup(evidence=not setup.args.cleanup)
+            except SetupFailure as cleanup_error:
+                setup.log(f"Cancellation cleanup failed: {cleanup_error}")
+        return 128 + exc.signum
+    except SetupFailure as exc:
+        print("[sql] " + redact(str(exc), password), file=sys.stderr, flush=True)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())