diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
deleted file mode 100644
index 5302abb5e..000000000
--- a/.pre-commit-config.yaml
+++ /dev/null
@@ -1,14 +0,0 @@
-repos:
-  - repo: https://github.com/psf/black
-    rev: 24.1.1
-    hooks:
-      - id: black
-        language_version: python3
-  - repo: https://github.com/PyCQA/isort
-    rev: 5.12.0
-    hooks:
-      - id: isort
-  - repo: https://github.com/PyCQA/flake8
-    rev: 7.0.0
-    hooks:
-      - id: flake8
diff --git a/eng/pack/templates/macos_64_env_gen.yml b/eng/pack/templates/macos_64_env_gen.yml
index b7ba0d56f..725d97127 100644
--- a/eng/pack/templates/macos_64_env_gen.yml
+++ b/eng/pack/templates/macos_64_env_gen.yml
@@ -99,7 +99,6 @@ steps:
       !*.dist-info/**
       !werkzeug/debug/shared/debugger.js
       !proxy_worker/**
-      !google/**
     targetFolder: '$(Build.ArtifactStagingDirectory)'
   condition: eq(variables['proxyWorker'], false)
   displayName: 'Copy azure_functions_worker files'
@@ -124,6 +123,7 @@ steps:
         "azure_functions_worker/protos/shared/NullableTypes_pb2.py",
         "azure_functions_worker/protos/shared/NullableTypes_pb2_grpc.py",
         "dateutil",
+        "google",
         "grpc",
         "markupsafe",
         "six.py",
diff --git a/eng/pack/templates/nix_arm64_env_gen.yml b/eng/pack/templates/nix_arm64_env_gen.yml
index 5e62912d2..b1993d98a 100644
--- a/eng/pack/templates/nix_arm64_env_gen.yml
+++ b/eng/pack/templates/nix_arm64_env_gen.yml
@@ -99,7 +99,6 @@ steps:
       !*.dist-info/**
       !werkzeug/debug/shared/debugger.js
       !proxy_worker/**
-      !google/**
     targetFolder: '$(Build.ArtifactStagingDirectory)'
   condition: eq(variables['proxyWorker'], false)
   displayName: 'Copy azure_functions_worker files'
@@ -124,6 +123,7 @@ steps:
         "azure_functions_worker/protos/shared/NullableTypes_pb2.py",
         "azure_functions_worker/protos/shared/NullableTypes_pb2_grpc.py",
         "dateutil",
+        "google",
         "grpc",
         "markupsafe",
         "six.py",
diff --git a/eng/pack/templates/nix_env_gen.yml b/eng/pack/templates/nix_env_gen.yml
index 1335cee26..468aecfd3 100644
--- a/eng/pack/templates/nix_env_gen.yml
+++ b/eng/pack/templates/nix_env_gen.yml
@@ -99,7 +99,6 @@ steps:
       !*.dist-info/**
       !werkzeug/debug/shared/debugger.js
       !proxy_worker/**
-      !google/**
     targetFolder: '$(Build.ArtifactStagingDirectory)'
   condition: eq(variables['proxyWorker'], false)
   displayName: 'Copy azure_functions_worker files'
@@ -124,6 +123,7 @@ steps:
         "azure_functions_worker/protos/shared/NullableTypes_pb2.py",
         "azure_functions_worker/protos/shared/NullableTypes_pb2_grpc.py",
         "dateutil",
+        "google",
         "grpc",
         "markupsafe",
         "six.py",
diff --git a/eng/pack/templates/win_env_gen.yml b/eng/pack/templates/win_env_gen.yml
index dbf15bfc0..32470a01a 100644
--- a/eng/pack/templates/win_env_gen.yml
+++ b/eng/pack/templates/win_env_gen.yml
@@ -98,7 +98,6 @@ steps:
       !*.dist-info\**
       !werkzeug\debug\shared\debugger.js
       !proxy_worker\**
-      !google\**
     targetFolder: '$(Build.ArtifactStagingDirectory)'
   condition: eq(variables['proxyWorker'], false)
   displayName: 'Copy azure_functions_worker files'
@@ -123,6 +122,7 @@ steps:
         "azure_functions_worker/protos/shared/NullableTypes_pb2.py",
         "azure_functions_worker/protos/shared/NullableTypes_pb2_grpc.py",
         "dateutil",
+        "google",
         "grpc",
         "markupsafe",
         "six.py",
diff --git a/eng/scripts/install-dependencies.sh b/eng/scripts/install-dependencies.sh
index 2ce956594..e7e0dd00c 100644
--- a/eng/scripts/install-dependencies.sh
+++ b/eng/scripts/install-dependencies.sh
@@ -1,15 +1,43 @@
 #!/bin/bash
+set -e
 
+# Install uv for faster dependency resolution / installation.
 python -m pip install --upgrade pip
-python -m pip install "setuptools>=62,<82.0"
-python -m pip install -e runtimes/v2
-python -m pip install -e runtimes/v1
-python -m pip install -U azure-functions --pre
-python -m pip install -U -e $2/[dev]
-
-python -m pip install --pre -U -e $2/[test-http-v2]
-python -m pip install --pre -U -e $2/[test-deferred-bindings]
-
-SERVICEBUS_DIR="./servicebus_dir"
-python -m pip install --pre -U --target "$SERVICEBUS_DIR" azurefunctions-extensions-bindings-servicebus==1.0.0b2
-python -c "import sys; sys.path.insert(0, '$SERVICEBUS_DIR'); import azurefunctions.extensions.bindings.servicebus as sb; print('servicebus version:', sb.__version__)"
+python -m pip install uv
+
+# Use uv as a drop-in replacement for pip. `--system` installs into the active
+# Python environment (the agent's Python), matching previous `pip install` behavior.
+UV_PIP="python -m uv pip install --system"
+
+# Setuptools is needed up front for editable installs of legacy packages.
+$UV_PIP "setuptools>=62,<82.0"
+
+# runtimes/v1 and runtimes/v2 require Python >= 3.13. Old pip would silently
+# install them on lower versions; uv (correctly) refuses, so install them
+# conditionally. They are only consumed by proxy_worker (Python >= 3.13).
+PY_VER="$1"
+PY_MINOR="${PY_VER#*.}"
+EXTRA_ARGS=()
+if [ "${PY_MINOR:-0}" -ge 13 ]; then
+    EXTRA_ARGS+=(-e runtimes/v2 -e runtimes/v1)
+fi
+
+# Install everything else in a single uv invocation so the resolver runs once
+# and all wheels are downloaded in parallel.
+$UV_PIP -U --prerelease=allow \
+    azure-functions \
+    -e "$2/[dev]" \
+    -e "$2/[test-http-v2]" \
+    -e "$2/[test-deferred-bindings]" \
+    "${EXTRA_ARGS[@]}"
+
+# The servicebus binding extension depends on uamqp, which is deprecated and
+# ships no wheels for Python 3.14 (source builds fail). Install it only on
+# Python < 3.14, mirroring the eventhub binding gate in pyproject.toml.
+if [ "${PY_MINOR:-0}" -lt 14 ]; then
+    SERVICEBUS_DIR="./servicebus_dir"
+    python -m uv pip install --prerelease=allow -U --target "$SERVICEBUS_DIR" azurefunctions-extensions-bindings-servicebus==1.0.0b2
+    python -c "import sys; sys.path.insert(0, '$SERVICEBUS_DIR'); import azurefunctions.extensions.bindings.servicebus as sb; print('servicebus version:', sb.__version__)"
+else
+    echo "Skipping servicebus binding extension on Python $PY_VER (uamqp has no 3.14 wheels)."
+fi
diff --git a/eng/scripts/test-extensions.sh b/eng/scripts/test-extensions.sh
index 2690f0d43..3bd5adeaf 100644
--- a/eng/scripts/test-extensions.sh
+++ b/eng/scripts/test-extensions.sh
@@ -1,10 +1,14 @@
 #!/bin/bash
+set -e
 
 python -m pip install --upgrade pip
+python -m pip install uv
 
-python -m pip install "setuptools>=62,<82.0"
-python -m pip install -e $1/PythonExtensionArtifact/$3
-python -m pip install --pre -e workers/[test-http-v2]
-python -m pip install --pre -U -e workers/[test-deferred-bindings]
+UV_PIP="python -m uv pip install --system"
 
-python -m pip install -U -e workers/[dev]
\ No newline at end of file
+$UV_PIP "setuptools>=62,<82.0"
+$UV_PIP -e $1/PythonExtensionArtifact/$3
+$UV_PIP --prerelease=allow -e workers/[test-http-v2]
+$UV_PIP -U --prerelease=allow -e workers/[test-deferred-bindings]
+
+$UV_PIP -U -e workers/[dev]
\ No newline at end of file
diff --git a/eng/scripts/test-sdk.sh b/eng/scripts/test-sdk.sh
index 900b1775e..cf5caf1e6 100644
--- a/eng/scripts/test-sdk.sh
+++ b/eng/scripts/test-sdk.sh
@@ -1,9 +1,14 @@
 #!/bin/bash
+set -e
 
 python -m pip install --upgrade pip
-python -m pip install "setuptools>=62,<82.0"
-python -m pip install -e $1/PythonSdkArtifact
-python -m pip install -e workers/[dev]
+python -m pip install uv
 
-python -m pip install --pre -U -e workers/[test-http-v2]
-python -m pip install --pre -U -e workers/[test-deferred-bindings]
\ No newline at end of file
+UV_PIP="python -m uv pip install --system"
+
+$UV_PIP "setuptools>=62,<82.0"
+$UV_PIP -e $1/PythonSdkArtifact
+$UV_PIP -e workers/[dev]
+
+$UV_PIP -U --prerelease=allow -e workers/[test-http-v2]
+$UV_PIP -U --prerelease=allow -e workers/[test-deferred-bindings]
\ No newline at end of file
diff --git a/eng/scripts/test-setup.sh b/eng/scripts/test-setup.sh
index f65d66e88..89ec2e4fa 100644
--- a/eng/scripts/test-setup.sh
+++ b/eng/scripts/test-setup.sh
@@ -1,4 +1,5 @@
 #!/bin/bash
+set -e
 
 cd workers/tests
 python -m invoke -c test_setup build-protos
diff --git a/eng/templates/jobs/ci-dependency-check.yml b/eng/templates/jobs/ci-dependency-check.yml
index 2334eb147..8a8521da9 100644
--- a/eng/templates/jobs/ci-dependency-check.yml
+++ b/eng/templates/jobs/ci-dependency-check.yml
@@ -55,8 +55,10 @@ jobs:
       - bash: |
             echo "Checking azure_functions_worker (Python < 3.13)..."
             cd workers
-            pip install "setuptools<82.0"
-            pip install . invoke
+            python -m pip install --upgrade pip
+            python -m pip install uv
+            python -m uv pip install --system "setuptools<82.0"
+            python -m uv pip install --system . invoke
             cd tests
             python -m invoke -c test_setup build-protos
             cd ..
@@ -66,7 +68,9 @@ jobs:
       - bash: |
             echo "Checking proxy_worker (Python >= 3.13)..."
             cd workers
-            pip install . invoke
+            python -m pip install --upgrade pip
+            python -m pip install uv
+            python -m uv pip install --system . invoke
             cd tests
             python -m invoke -c test_setup build-protos
             cd ..
@@ -76,14 +80,18 @@ jobs:
       - bash: |
             echo "Checking V1 Library Worker (Python >= 3.13)..."
             cd runtimes/v1
-            pip install .
+            python -m pip install --upgrade pip
+            python -m pip install uv
+            python -m uv pip install --system .
             python -c "import pkgutil, importlib; [importlib.import_module(f'azure_functions_runtime_v1.{name}') for _, name, _ in pkgutil.walk_packages(['azure_functions_runtime_v1'])]"
         displayName: 'Python Library V1: check for missing dependencies'
         condition: eq(variables['proxyWorker'], true)
       - bash: |
             echo "Checking V2 Library Worker (Python >= 3.13)..."
             cd runtimes/v2
-            pip install .
+            python -m pip install --upgrade pip
+            python -m pip install uv
+            python -m uv pip install --system .
             python -c "import pkgutil, importlib; [importlib.import_module(f'azure_functions_runtime.{name}') for _, name, _ in pkgutil.walk_packages(['azure_functions_runtime'])]"
         displayName: 'Python Library V2: check for missing dependencies'
         condition: eq(variables['proxyWorker'], true)
diff --git a/eng/templates/jobs/ci-emulator-tests.yml b/eng/templates/jobs/ci-emulator-tests.yml
index 45107c846..7aeb3b24f 100644
--- a/eng/templates/jobs/ci-emulator-tests.yml
+++ b/eng/templates/jobs/ci-emulator-tests.yml
@@ -72,11 +72,13 @@ jobs:
         displayName: 'Configure NuGet feed for current organization'
       - bash: |
           chmod +x eng/scripts/install-dependencies.sh
-          chmod +x eng/scripts/test-setup.sh
-
           eng/scripts/install-dependencies.sh $(PYTHON_VERSION) ${{ parameters.PROJECT_DIRECTORY }}
+        displayName: 'Install Python dependencies (uv)'
+        condition: and(eq(variables.isSdkRelease, false), eq(variables.isExtensionsRelease, false), eq(variables['USETESTPYTHONSDK'], false), eq(variables['USETESTPYTHONEXTENSIONS'], false))
+      - bash: |
+          chmod +x eng/scripts/test-setup.sh
           eng/scripts/test-setup.sh
-        displayName: 'Install dependencies and the worker'
+        displayName: 'Build webhost and extensions (dotnet)'
         condition: and(eq(variables.isSdkRelease, false), eq(variables.isExtensionsRelease, false), eq(variables['USETESTPYTHONSDK'], false), eq(variables['USETESTPYTHONEXTENSIONS'], false))
       - task: DownloadPipelineArtifact@2
         displayName: 'Download Python SDK Artifact'
@@ -128,7 +130,7 @@ jobs:
           docker compose -f ${{ parameters.PROJECT_DIRECTORY }}/tests/emulator_tests/utils/eventhub/docker-compose.yml up -d
         displayName: 'Install Azurite and Start EventHub Emulator'
       - bash: |
-            python -m pytest -q --dist loadfile --reruns 4 --ignore=tests/emulator_tests/test_servicebus_functions.py tests/emulator_tests
+            python -m pytest -q -n 2 --dist loadfile --reruns 4 --ignore=tests/emulator_tests/test_servicebus_functions.py tests/emulator_tests
         env:
           AzureWebJobsStorage: $(AzureWebJobsStorage)
           AZURE_STORAGE_CONNECTION_STRING: $(AZURE_STORAGE_CONNECTION_STRING)
@@ -144,6 +146,24 @@ jobs:
           docker container rm --force eventhubs-emulator
           docker compose -f ${{ parameters.PROJECT_DIRECTORY }}/tests/emulator_tests/utils/servicebus/docker-compose.yml pull
           docker compose -f ${{ parameters.PROJECT_DIRECTORY }}/tests/emulator_tests/utils/servicebus/docker-compose.yml up -d
+
+          # Wait for the Service Bus emulator (and its SQL Edge dependency) to
+          # be ready before running tests. Starting the webhost while the
+          # emulator is still booting makes the ServiceBus listener fail at
+          # startup, so the host never binds and tests see connection refused.
+          echo "Waiting for the Service Bus emulator to be ready..."
+          for i in $(seq 1 90); do
+            if docker logs servicebus-emulator 2>&1 | grep -q "Emulator Service is Successfully Up"; then
+              echo "Service Bus emulator is ready."
+              break
+            fi
+            if [ "$i" -eq 90 ]; then
+              echo "Service Bus emulator did not become ready in time. Logs:"
+              docker logs servicebus-emulator || true
+              exit 1
+            fi
+            sleep 2
+          done
         env:
           AzureWebJobsSQLPassword: $(AzureWebJobsSQLPassword)
         displayName: 'Install Azurite and Start ServiceBus Emulator'
diff --git a/eng/templates/jobs/ci-library-unit-tests.yml b/eng/templates/jobs/ci-library-unit-tests.yml
index 9b25e6a61..74091dc17 100644
--- a/eng/templates/jobs/ci-library-unit-tests.yml
+++ b/eng/templates/jobs/ci-library-unit-tests.yml
@@ -49,7 +49,7 @@ jobs:
           eng/scripts/install-dependencies.sh $(PYTHON_VERSION) ${{ parameters.PROJECT_DIRECTORY }}
         displayName: 'Install dependencies'
       - bash: |
-          python -m pytest -q --dist loadfile --reruns 4 --instafail --cov=./${{ parameters.PROJECT_DIRECTORY }} --cov-report xml --cov-branch tests/unittests
+          python -m pytest -q -n auto --dist loadfile --reruns 4 --reruns-delay 5 --instafail --cov=./${{ parameters.PROJECT_DIRECTORY }} --cov-report xml --cov-branch tests/unittests
         displayName: "Running $(PYTHON_VERSION) Unit Tests"
         env:
           AZURE_STORAGE_CONNECTION_STRING: $(AZURE_STORAGE_CONNECTION_STRING)
diff --git a/eng/templates/jobs/ci-unit-tests.yml b/eng/templates/jobs/ci-unit-tests.yml
index 798bedc1b..5736307e5 100644
--- a/eng/templates/jobs/ci-unit-tests.yml
+++ b/eng/templates/jobs/ci-unit-tests.yml
@@ -72,11 +72,13 @@ jobs:
         displayName: 'Configure NuGet feed for current organization'
       - bash: |
           chmod +x eng/scripts/install-dependencies.sh
-          chmod +x eng/scripts/test-setup.sh
-          
           eng/scripts/install-dependencies.sh $(PYTHON_VERSION) ${{ parameters.PROJECT_DIRECTORY }}
+        displayName: 'Install Python dependencies (uv)'
+        condition: and(eq(variables.isSdkRelease, false), eq(variables.isExtensionsRelease, false), eq(variables['USETESTPYTHONSDK'], false), eq(variables['USETESTPYTHONEXTENSIONS'], false))
+      - bash: |
+          chmod +x eng/scripts/test-setup.sh
           eng/scripts/test-setup.sh
-        displayName: 'Install dependencies'
+        displayName: 'Build webhost and extensions (dotnet)'
         condition: and(eq(variables.isSdkRelease, false), eq(variables.isExtensionsRelease, false), eq(variables['USETESTPYTHONSDK'], false), eq(variables['USETESTPYTHONEXTENSIONS'], false))
       - bash: |
           PY_VER="$(PYTHON_VERSION)"
@@ -87,11 +89,11 @@ jobs:
 
           if [ "$PY_MINOR" -ge 13 ]; then
               echo "Running proxy_worker tests (Python >= 3.13)..."
-              python -m pytest -q --dist loadfile --reruns 4 --instafail \
+              python -m pytest -q -n auto --dist loadfile --reruns 4 --reruns-delay 5 --instafail \
                     --cov=./proxy_worker --cov-report xml --cov-branch tests/unittest_proxy
           else
               echo "Running unittests (Python < 3.13)..."
-              python -m pytest -q --dist loadfile --reruns 4 --instafail \
+              python -m pytest -q -n auto --dist loadfile --reruns 4 --reruns-delay 5 --instafail \
                     --cov=./azure_functions_worker --cov-report xml --cov-branch tests/unittests
           fi
         displayName: "Running $(PYTHON_VERSION) Unit Tests"
diff --git a/eng/templates/official/jobs/ci-e2e-tests.yml b/eng/templates/official/jobs/ci-e2e-tests.yml
index 85c759a9b..aec4f667b 100644
--- a/eng/templates/official/jobs/ci-e2e-tests.yml
+++ b/eng/templates/official/jobs/ci-e2e-tests.yml
@@ -95,11 +95,13 @@ jobs:
         displayName: 'Configure NuGet feed for current organization'
       - bash: |
           chmod +x eng/scripts/install-dependencies.sh
-          chmod +x eng/scripts/test-setup.sh
-          
           eng/scripts/install-dependencies.sh $(PYTHON_VERSION) ${{ parameters.PROJECT_DIRECTORY }}
+        displayName: 'Install Python dependencies (uv)'
+        condition: and(eq(variables.isSdkRelease, false), eq(variables.isExtensionsRelease, false), eq(variables['USETESTPYTHONSDK'], false), eq(variables['USETESTPYTHONEXTENSIONS'], false))
+      - bash: |
+          chmod +x eng/scripts/test-setup.sh
           eng/scripts/test-setup.sh
-        displayName: 'Install dependencies and the worker'
+        displayName: 'Build webhost and extensions (dotnet)'
         condition: and(eq(variables.isSdkRelease, false), eq(variables.isExtensionsRelease, false), eq(variables['USETESTPYTHONSDK'], false), eq(variables['USETESTPYTHONEXTENSIONS'], false))
       - task: DownloadPipelineArtifact@2
         displayName: 'Download Python SDK Artifact'
diff --git a/eng/templates/official/jobs/ci-fc-tests.yml b/eng/templates/official/jobs/ci-fc-tests.yml
index 41e394a0c..7b4484ada 100644
--- a/eng/templates/official/jobs/ci-fc-tests.yml
+++ b/eng/templates/official/jobs/ci-fc-tests.yml
@@ -66,8 +66,10 @@ jobs:
 
       - bash: |
           python -m pip install --upgrade pip
-          pip install "setuptools<82.0"
-          python -m pip install -U -e ${{ parameters.PROJECT_DIRECTORY }}/[dev]
+          python -m pip install uv
+          UV_PIP="python -m uv pip install --system"
+          $UV_PIP "setuptools<82.0"
+          $UV_PIP -U -e ${{ parameters.PROJECT_DIRECTORY }}/[dev]
 
           cd ${{ parameters.PROJECT_DIRECTORY }}/tests
           python -m invoke -c test_setup build-protos
diff --git a/runtimes/v1/pyproject.toml b/runtimes/v1/pyproject.toml
index d8b474e56..d074d1113 100644
--- a/runtimes/v1/pyproject.toml
+++ b/runtimes/v1/pyproject.toml
@@ -40,14 +40,11 @@ dev = [
     "grpcio~=1.70.0",
     "grpcio-tools~=1.70.0",
     "protobuf~=5.29.0",
-    "pytest-sugar",
     "pytest-cov",
     "pytest-xdist",
-    "pytest-randomly",
     "pytest-instafail",
     "pytest-rerunfailures",
     "pytest-asyncio",
-    "pre-commit",
     "invoke"
 ]
 
diff --git a/runtimes/v2/pyproject.toml b/runtimes/v2/pyproject.toml
index 2e83920d8..d58a574b9 100644
--- a/runtimes/v2/pyproject.toml
+++ b/runtimes/v2/pyproject.toml
@@ -42,14 +42,11 @@ dev = [
     "grpcio~=1.70.0",
     "grpcio-tools~=1.70.0",
     "protobuf~=5.29.0",
-    "pytest-sugar",
     "pytest-cov",
     "pytest-xdist",
-    "pytest-randomly",
     "pytest-instafail",
     "pytest-rerunfailures",
     "pytest-asyncio",
-    "pre-commit",
     "invoke"
 ]
 test-http-v2 = [
diff --git a/runtimes/v2/tests/unittests/test_deferred_bindings.py b/runtimes/v2/tests/unittests/test_deferred_bindings.py
index 66f40c3bf..f82773c12 100644
--- a/runtimes/v2/tests/unittests/test_deferred_bindings.py
+++ b/runtimes/v2/tests/unittests/test_deferred_bindings.py
@@ -10,7 +10,6 @@
 
 
 from azurefunctions.extensions.bindings.blob import (BlobClient,
-                                                     BlobClientConverter,
                                                      ContainerClient,
                                                      StorageStreamDownloader)
 from azurefunctions.extensions.bindings.eventhub import EventData, EventDataConverter
@@ -24,27 +23,6 @@ def setUp(self):
         # Initialize DEFERRED_BINDING_REGISTRY
         meta.load_binding_registry()
 
-    def test_mbd_deferred_bindings_enabled_decode(self):
-        binding = BlobClientConverter
-        pb = protos.ParameterBinding(name='test',
-                                     data=protos.TypedData(
-                                         string='test'))
-        sample_mbd = MockMBD(version="1.0",
-                             source="AzureStorageBlobs",
-                             content_type="application/json",
-                             content="{\"Connection\":\"AZURE_STORAGE_CONNECTION_STRING\","  # noqa
-                                     "\"ContainerName\":"
-                                     "\"python-worker-tests\","
-                                     "\"BlobName\":"
-                                     "\"test-blobclient-trigger.txt\"}")
-        datum = datumdef.Datum(value=sample_mbd, type='model_binding_data')
-
-        obj = meta.deferred_bindings_decode(binding=binding, pb=pb,
-                                            pytype=BlobClient, datum=datum, metadata={},
-                                            function_name="test_function")
-
-        self.assertIsNotNone(obj)
-
     def test_cmbd_deferred_bindings_enabled_decode(self):
         binding = EventDataConverter
         pb = protos.ParameterBinding(name='test',
diff --git a/workers/azure_functions_worker/__init__.py b/workers/azure_functions_worker/__init__.py
index 4f7aba94a..25c6350a3 100644
--- a/workers/azure_functions_worker/__init__.py
+++ b/workers/azure_functions_worker/__init__.py
@@ -1,120 +1,136 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
+import importlib.util
 import os
+import re
 import sys
 
-# ---------------------------------------------------------------------------
 # Protobuf runtime selection
-# ---------------------------------------------------------------------------
 #
-# The worker's generated ``*_pb2.py`` stubs import ``google.protobuf``
-# at the top level. Two scenarios:
+# The worker's generated ``*_pb2.py`` stubs, loader and converters all
+# import top-level ``google.protobuf``, so the whole worker shares one
+# protobuf runtime and descriptor pool. Which protobuf that is depends on
+# what is first on ``sys.path`` (the function app's ``.python_packages`` and
+# PYTHONPATH precede the worker's own protobuf):
 #
-# 1. The function app does NOT ship its own ``google.protobuf``. The
-#    top-level lookup resolves to the protobuf install that ships with
-#    the worker runtime (under ``worker_deps_path`` on Azure Functions),
-#    which is guaranteed compatible with the worker's pb2 stubs and
-#    includes the fast ``upb`` C extension. Nothing to do here.
+# 1. The protobuf on the path is the same or newer than the worker's
+#    vendored copy (including the worker's own protobuf, or protobuf 6.x
+#    shipped by an extension): use it directly. The worker's stubs run fine
+#    on a same-or-newer runtime.
 #
-# 2. The function app DOES ship ``google.protobuf`` in
-#    ``.python_packages``. On Azure Functions the customer's path
-#    precedes the worker's on ``sys.path``, so a top-level
-#    ``import google.protobuf`` resolves to the customer's copy. If the
-#    customer pinned an older protobuf (the common case is 4.x) the
-#    worker's pb2 stubs fail to load — for example ``from
-#    google.protobuf import runtime_version`` does not exist before
-#    protobuf 5.27. To insulate the worker from the customer's pin we:
-#       a. Pre-import the vendored ``google.protobuf`` modules and
-#          register them in ``sys.modules`` under their top-level names
-#          so subsequent ``from google.protobuf import X`` resolves to
-#          the vendored copy.
-#       b. Force the vendored copy onto its pure-Python implementation
-#          via ``PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python`` so the
-#          vendored ``api_implementation`` does not try to load the
-#          customer's ``google._upb._message`` C extension (which would
-#          be incompatible with vendored protobuf and unsafe to load
-#          alongside another ``_upb`` instance).
+# 2. The protobuf on the path is older than the vendored copy (e.g. an app
+#    pins protobuf 4.x): alias top-level ``google.protobuf`` to the vendored
+#    copy in ``sys.modules`` and force pure-Python
+#    (``PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python``), so the worker's
+#    stubs do not load against an incompatible protobuf. The app's older
+#    protobuf is shadowed for the whole process.
 #
-# Side effect of scenario 2: customer code that does ``import
-# google.protobuf`` later in the process will resolve to the vendored
-# copy rather than the customer's pinned copy. This trade-off is
-# necessary because protobuf's runtime assumes a single coherent
-# ``google.protobuf`` package per process.
-#
-# Detection cost: a single env-var lookup plus at most one
-# ``os.path.isdir`` call at worker startup. Zero per invocation.
-#
-# Policy override via ``_AZFUNC_USE_VENDORED_PROTOBUF``:
-#   ``"1"`` — force activation. The launcher (``worker.py``) sets this
-#            in local-dev mode so we always isolate the worker from
-#            whatever protobuf version sits in the customer's venv.
-#   ``"0"`` — force no activation. Escape hatch for users who need to
-#            debug protobuf-version-specific behavior against the
-#            worker's bundled protobuf.
-#   unset   — autodetect via the canonical Azure Functions layout
-#            (``.python_packages``). This is the production path; the
-#            override env var is not set in cloud launches.
+# Override via ``_AZFUNC_USE_VENDORED_PROTOBUF``: ``"1"`` forces activation,
+# ``"0"`` forces off, unset autodetects by comparing versions.
 
 _USE_VENDORED_PROTOBUF_ENV = "_AZFUNC_USE_VENDORED_PROTOBUF"
 
 
+def _parse_protobuf_version(version_str):
+    """Parse a protobuf version string into a comparable tuple of ints.
+
+    Only the leading numeric dotted components are used; any pre-release
+    or local suffix (e.g. ``rc1``) is ignored. Returns ``None`` if no
+    numeric version can be parsed.
+    """
+    parts = []
+    for token in version_str.split('.'):
+        match = re.match(r'\d+', token.strip())
+        if not match:
+            break
+        parts.append(int(match.group()))
+    return tuple(parts) if parts else None
+
+
+def _read_protobuf_version(protobuf_dir):
+    """Read ``__version__`` from a ``google/protobuf`` package directory
+    without importing it. Returns a comparable version tuple or ``None``
+    when it cannot be determined.
+    """
+    init_path = os.path.join(protobuf_dir, "__init__.py")
+    try:
+        with open(init_path, "r", encoding="utf-8") as f:
+            content = f.read()
+    except OSError:
+        return None
+    match = re.search(
+        r"__version__\s*=\s*['\"]([^'\"]+)['\"]", content)
+    if not match:
+        return None
+    return _parse_protobuf_version(match.group(1))
+
+
+def _vendored_protobuf_dir():
+    return os.path.join(
+        os.path.dirname(__file__),
+        "_vendored", "google", "protobuf",
+    )
+
+
+def _find_importable_protobuf_dir():
+    """Return the directory of the top-level ``google.protobuf`` that would
+    be imported from ``sys.path``, without importing it. None if not found.
+    """
+    try:
+        spec = importlib.util.find_spec("google.protobuf")
+    except (ImportError, ValueError, AttributeError):
+        return None
+    if spec is None or not spec.origin:
+        return None
+    return os.path.dirname(spec.origin)
+
+
 def _should_use_vendored_protobuf() -> bool:
     """Return True if the worker should activate its private pure-Python
     ``google.protobuf`` fallback for this process.
 
-    The launcher (``worker.py``) is the policy layer: it knows whether
-    we are running in Azure or locally and sets
-    ``_AZFUNC_USE_VENDORED_PROTOBUF`` accordingly. If the env var is
-    unset (e.g. the worker was imported directly by a test or a
-    third-party host) we fall back to checking the canonical Azure
-    Functions deployment layout.
-
-    We deliberately do *not* use a generic ``importlib.util.find_spec``
-    lookup as a fallback because that would also match the worker's
-    own protobuf install (which is always on ``sys.path`` and is not
-    "customer protobuf"). A false positive there would activate the
-    pure-Python vendored fallback for every function app and erase
-    the perf benefit of running the worker on ``upb``.
+    ``_AZFUNC_USE_VENDORED_PROTOBUF`` forces the choice when set. Otherwise
+    we look at the ``google.protobuf`` that would be imported from
+    ``sys.path`` and fall back to the vendored copy only when it is older
+    than ours; a same-or-newer protobuf (including the worker's own) is used
+    directly, so protobuf-6 extensions load and the worker keeps ``upb``.
     """
     override = os.environ.get(_USE_VENDORED_PROTOBUF_ENV)
     if override == "1":
         return True
     if override == "0":
         return False
-    script_root = os.environ.get("AzureWebJobsScriptRoot")
-    if not script_root:
+
+    protobuf_dir = _find_importable_protobuf_dir()
+    if protobuf_dir is None:
         return False
-    candidate = os.path.join(
-        script_root,
-        ".python_packages",
-        "lib",
-        "site-packages",
-        "google",
-        "protobuf",
-    )
-    return os.path.isdir(candidate)
+    if "_vendored" in protobuf_dir.split(os.sep):
+        # Already resolves to our vendored copy.
+        return False
+
+    app_version = _read_protobuf_version(protobuf_dir)
+    vendored_version = _read_protobuf_version(_vendored_protobuf_dir())
+    if app_version is None or vendored_version is None:
+        # Cannot compare versions; insulate the worker (old behavior).
+        return True
+    # Fall back to vendored only when the path protobuf is older than ours.
+    return app_version < vendored_version
 
 
 def _activate_vendored_protobuf() -> None:
-    """Pre-import the vendored protobuf modules and alias them under
-    the top-level ``google.protobuf`` names so the worker's pb2 stubs
-    resolve to the vendored copy instead of the customer's pinned one.
+    """Pre-import the vendored protobuf modules and alias them under the
+    top-level ``google.protobuf`` names so the worker's pb2 stubs resolve to
+    the vendored copy instead of the function app's.
     """
     try:
         import importlib
 
-        # Alias only the protobuf-specific names. Do NOT alias the
-        # top-level ``google`` package: the vendored ``google`` is a
-        # regular package whose ``__path__`` covers only our vendored
-        # tree, so aliasing it would shadow every other ``google.*``
-        # the customer ships (``google.cloud.*``, ``google.auth``,
-        # ``google.api_core``, etc.). Those packages are the most
-        # common reason a customer ends up with protobuf in their
-        # dependencies in the first place, so breaking them would
-        # defeat the purpose of the fallback. ``from google.protobuf
-        # import X`` short-circuits on ``sys.modules["google.protobuf"]``
-        # without consulting ``sys.modules["google"]``, so aliasing
-        # only the leaves is sufficient.
+        # Alias only the protobuf-specific names, not the top-level
+        # ``google`` package: the vendored ``google`` covers only our tree,
+        # so aliasing it would shadow other ``google.*`` the function app
+        # ships (``google.cloud.*``, ``google.auth``, etc.). ``from
+        # google.protobuf import X`` short-circuits on
+        # ``sys.modules["google.protobuf"]``, so aliasing the leaves is enough.
         modules_to_alias = (
             "google.protobuf",
             "google.protobuf.internal",
@@ -122,32 +138,25 @@ def _activate_vendored_protobuf() -> None:
         for top_name in modules_to_alias:
             vendored_name = "azure_functions_worker._vendored." + top_name
             mod = importlib.import_module(vendored_name)
-            # Force the alias even if something already populated
-            # ``sys.modules`` for the top-level name. The whole point
-            # of activation is "the customer's protobuf must not be
-            # what the worker's pb2 stubs see"; ``setdefault`` would
-            # let an early customer import keep the slot.
+            # Force the alias even if the name is already in ``sys.modules``;
+            # ``setdefault`` would let an early import keep the slot.
             sys.modules[top_name] = mod
     except ImportError:
         # Vendored tree may be absent in some dev workflows (before
-        # ``vendor_deps.py`` has been run). Stay quiet here; the next
-        # worker import will surface a clearer error.
+        # ``vendor_deps.py`` runs). Stay quiet; a later import surfaces it.
         return
 
 
 if _should_use_vendored_protobuf():
-    # Force the vendored copy onto pure-Python BEFORE pre-importing
-    # any of its modules, so that vendored ``api_implementation``
-    # doesn't try to load a (potentially incompatible) ``_upb``.
+    # Force pure-Python before importing the vendored modules so its
+    # ``api_implementation`` does not load an incompatible ``_upb``.
     os.environ.setdefault(
         "PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION", "python"
     )
     _activate_vendored_protobuf()
-# else: nothing to do. Worker's pb2 stubs will resolve top-level
-# google.protobuf to the worker's own protobuf install and use upb
-# naturally. We deliberately do NOT log on the no-op path: it would
-# run on every worker startup for the entire fleet and provides no
-# actionable signal to customers.
+# else: nothing to do. Stubs resolve top-level google.protobuf to the
+# worker's own protobuf and use upb. No log on this path; it runs on every
+# startup and gives no actionable signal.
 
 
 del _should_use_vendored_protobuf
diff --git a/workers/pyproject.toml b/workers/pyproject.toml
index deacfb95d..b1a4b2a96 100644
--- a/workers/pyproject.toml
+++ b/workers/pyproject.toml
@@ -54,29 +54,25 @@ dev = [
     "azure-monitor-opentelemetry", # Used for Azure Monitor unit tests
     "azure-storage-blob~=12.27.1", # Used for Blob Emulator tests
     "flask",
-    "fastapi~=0.103.2",
+    "fastapi",
     "pydantic",
     "flake8==6.*",
     "mypy",
     "pytest~=7.4.4",
     "requests==2.*",
     "coverage",
-    "pytest-sugar",
     "opentelemetry-api", # Used for OpenTelemetry unit tests
     "pytest-cov",
     "pytest-xdist",
-    "pytest-randomly",
     "pytest-instafail",
     "pytest-rerunfailures",
     "pytest-asyncio",
-    "ptvsd",
     "python-dotenv",
     "plotly",
     "scikit-learn",
     "opencv-python",
     "pandas",
     "numpy",
-    "pre-commit",
     "invoke",
     "cryptography",
     "pyjwt",
diff --git a/workers/python/prodV4/worker.py b/workers/python/prodV4/worker.py
index fe0d26e4c..129b7ecf6 100644
--- a/workers/python/prodV4/worker.py
+++ b/workers/python/prodV4/worker.py
@@ -66,11 +66,10 @@ def determine_user_pkg_paths():
         # third-party user packages (in .venv)
         sys.path.insert(1, func_worker_dir)
         add_script_root_to_sys_path()
-        # In local dev we always activate the worker's private
-        # pure-Python ``google.protobuf`` fallback so the worker is
-        # isolated from whatever protobuf version sits in the
-        # user's environment
-        os.environ.setdefault("_AZFUNC_USE_VENDORED_PROTOBUF", "1")
+        # Let the worker autodetect protobuf: it falls back to the vendored
+        # pure-Python google.protobuf only when the function app ships an
+        # older protobuf, and otherwise uses the app's protobuf (e.g. 6.x
+        # from an extension).
         from azure_functions_worker import main
 
         main.main()
diff --git a/workers/python/test/worker.py b/workers/python/test/worker.py
index 33a987a08..8a606c8de 100644
--- a/workers/python/test/worker.py
+++ b/workers/python/test/worker.py
@@ -17,11 +17,10 @@ def add_script_root_to_sys_path():
     add_script_root_to_sys_path()
     minor_version = sys.version_info[1]
     if minor_version < 13:
-        # Local/test launch of the azure_functions_worker. Activate
-        # the worker's private pure-Python google.protobuf fallback
-        # so the worker is isolated from whatever protobuf version
-        # sits in the local environment.
-        os.environ.setdefault("_AZFUNC_USE_VENDORED_PROTOBUF", "1")
+        # Local/test launch of the azure_functions_worker. Let the worker
+        # autodetect protobuf: it falls back to the vendored pure-Python
+        # google.protobuf only when the function app ships an older protobuf,
+        # and otherwise uses the app's protobuf (e.g. 6.x from an extension).
         from azure_functions_worker import main
         main.main()
     else:
diff --git a/workers/tests/consumption_tests/function_app_zips/BlobSdkBindings.zip b/workers/tests/consumption_tests/function_app_zips/BlobSdkBindings.zip
new file mode 100644
index 000000000..700171e31
Binary files /dev/null and b/workers/tests/consumption_tests/function_app_zips/BlobSdkBindings.zip differ
diff --git a/workers/tests/consumption_tests/test_flex_consumption.py b/workers/tests/consumption_tests/test_flex_consumption.py
index b3db39a8f..e20a629ef 100644
--- a/workers/tests/consumption_tests/test_flex_consumption.py
+++ b/workers/tests/consumption_tests/test_flex_consumption.py
@@ -160,6 +160,27 @@ def test_pinning_functions_to_older_version(self):
             self.assertEqual(resp.status_code, 200)
             self.assertIn("Func Version: 1.11.1", resp.text)
 
+    @skipIf(sys.version_info.minor != 14,
+            "The BlobSdkBindings fixture bundles binary dependencies "
+            "(cryptography, cffi) built for Python 3.14.")
+    def test_blob_sdk_bindings(self):
+        """A function app using the azurefunctions blob SDK (deferred)
+        bindings extension should index and serve requests under Flex
+        Consumption, confirming SDK bindings work with Flex.
+        """
+        with FlexConsumptionWebHostController(_DEFAULT_HOST_VERSION,
+                                              self._py_version) as ctrl:
+            ctrl.assign_container(env={
+                "AzureWebJobsStorage": self._storage,
+                "SCM_RUN_FROM_PACKAGE": self._get_function_app(
+                    "BlobSdkBindings"),
+            })
+            req = Request('GET', f'{ctrl.url}/api/sdk_bindings_check')
+            resp = ctrl.send_request(req)
+
+            self.assertEqual(resp.status_code, 200)
+            self.assertIn("BlobClient", resp.text)
+
     @skipIf(
         sys.version_info >= (3, 14),
         "Opencensus bundles protobuf 4.24.0, which generates message "
diff --git a/workers/tests/emulator_tests/test_deferred_bindings_blob_functions.py b/workers/tests/emulator_tests/test_deferred_bindings_blob_functions.py
index 47b3801b8..524f4ffbd 100644
--- a/workers/tests/emulator_tests/test_deferred_bindings_blob_functions.py
+++ b/workers/tests/emulator_tests/test_deferred_bindings_blob_functions.py
@@ -16,6 +16,23 @@ def get_script_dir(cls):
     def get_libraries_to_install(cls):
         return ['azurefunctions-extensions-bindings-blob']
 
+    def test_mbd_deferred_bindings_enabled_decode(self):
+        # Moved from the runtimes/v2 unit test of the same name (which only
+        # asserted deferred_bindings_decode returned a BlobClient). Here the
+        # same decode path is exercised end-to-end against the storage
+        # emulator: a blob is written, then read back through a function whose
+        # input is typed as the BlobClient SDK type, confirming the model
+        # binding data was decoded into a usable BlobClient.
+        r = self.webhost.request('POST', 'put_blob_str', data='test-data')
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.text, 'OK')
+
+        time.sleep(5)
+
+        r = self.webhost.request('GET', 'get_bc_str')
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.text, 'test-data')
+
     def test_blob_str(self):
         r = self.webhost.request('POST', 'put_blob_str', data='test-data')
         self.assertEqual(r.status_code, 200)
diff --git a/workers/tests/emulator_tests/utils/eventhub/docker-compose.yml b/workers/tests/emulator_tests/utils/eventhub/docker-compose.yml
index 2c40aa042..068b0d93f 100644
--- a/workers/tests/emulator_tests/utils/eventhub/docker-compose.yml
+++ b/workers/tests/emulator_tests/utils/eventhub/docker-compose.yml
@@ -22,6 +22,7 @@ services:
   azurite:
     container_name: "azurite"
     image: "mcr.microsoft.com/azure-storage/azurite:latest"
+    command: "azurite --blobHost 0.0.0.0 --queueHost 0.0.0.0 --tableHost 0.0.0.0 --skipApiVersionCheck --loose"
     ports:
       - "10000:10000"
       - "10001:10001"
diff --git a/workers/tests/emulator_tests/utils/servicebus/docker-compose.yml b/workers/tests/emulator_tests/utils/servicebus/docker-compose.yml
index c1781a858..0fe5b5694 100644
--- a/workers/tests/emulator_tests/utils/servicebus/docker-compose.yml
+++ b/workers/tests/emulator_tests/utils/servicebus/docker-compose.yml
@@ -32,6 +32,7 @@ services:
   azurite:
     container_name: "azurite-sb"
     image: "mcr.microsoft.com/azure-storage/azurite:latest"
+    command: "azurite --blobHost 0.0.0.0 --queueHost 0.0.0.0 --tableHost 0.0.0.0 --skipApiVersionCheck --loose"
     ports:
       - "10003:10003"
       - "10004:10004"
diff --git a/workers/tests/test_setup.py b/workers/tests/test_setup.py
index 59f6fb7c6..2f785dcb7 100644
--- a/workers/tests/test_setup.py
+++ b/workers/tests/test_setup.py
@@ -115,10 +115,19 @@ def chmod_protobuf_generation_script(webhost_dir):
 
 def compile_webhost(webhost_dir):
     print(f"Compiling Functions Host from {webhost_dir}")
+    # Build only the WebHost project (and its dependencies) instead of the
+    # entire WebJobs.Script.sln. The solution also contains test projects,
+    # benchmarks and isolated-worker samples that the tests never run; building
+    # them is slow, consumes far more disk, and pulls many extra NuGet packages
+    # that can fail to restore on the internal CI feed. The WebHost project
+    # output already contains the full runtime dependency closure needed to run
+    # the host.
+    webhost_project = (pathlib.Path("src") / "WebJobs.Script.WebHost"
+                       / "WebJobs.Script.WebHost.csproj")
     try:
         subprocess.run(
             [
-                "dotnet", "build", "WebJobs.Script.sln",
+                "dotnet", "build", str(webhost_project),
                 "/m:1",  # Disable parallel MSBuild
                 "/nodeReuse:false",  # Prevent MSBuild node reuse
                 f"--property:OutputPath={webhost_dir}/bin",  # Set output folder
diff --git a/workers/tests/unittests/test_http_functions.py b/workers/tests/unittests/test_http_functions.py
index b9bac60ac..4d90fa19d 100644
--- a/workers/tests/unittests/test_http_functions.py
+++ b/workers/tests/unittests/test_http_functions.py
@@ -202,9 +202,14 @@ def test_accept_json(self):
 
         self.assertIn('accept_json', req['url'])
 
+    @testutils.retryable_test(3, 5)
     def test_unhandled_error(self):
         r = self.webhost.request('GET', 'unhandled_error')
         self.assertEqual(r.status_code, 500)
+        # The worker forwards the traceback to the host asynchronously, so
+        # wait for it to land in the host log before check_log_unhandled_error
+        # reads its snapshot.
+        self.wait_for_host_log('ZeroDivisionError: division by zero')
         # https://github.com/Azure/azure-functions-host/issues/2706
         # self.assertIn('Exception: ZeroDivisionError', r.text)
 
diff --git a/workers/tests/unittests/test_http_functions_v2.py b/workers/tests/unittests/test_http_functions_v2.py
index ce0c7f3ee..eba58d444 100644
--- a/workers/tests/unittests/test_http_functions_v2.py
+++ b/workers/tests/unittests/test_http_functions_v2.py
@@ -206,6 +206,10 @@ def test_accept_json(self):
     def test_unhandled_error(self):
         r = self.webhost.request('GET', 'unhandled_error')
         self.assertEqual(r.status_code, 500)
+        # The worker forwards the traceback to the host asynchronously, so
+        # wait for it to land in the host log before check_log_unhandled_error
+        # reads its snapshot.
+        self.wait_for_host_log('ZeroDivisionError: division by zero')
         # https://github.com/Azure/azure-functions-host/issues/2706
         # self.assertIn('Exception: ZeroDivisionError', r.text)
 
diff --git a/workers/tests/unittests/test_third_party_http_functions.py b/workers/tests/unittests/test_third_party_http_functions.py
index e64c7cb6a..84baafbe9 100644
--- a/workers/tests/unittests/test_third_party_http_functions.py
+++ b/workers/tests/unittests/test_third_party_http_functions.py
@@ -134,9 +134,13 @@ def test_return_http_no_body(self):
             self.assertEqual(r.text, '')
             self.assertEqual(r.status_code, 200)
 
+        @testutils.retryable_test(3, 5)
         def test_unhandled_error(self):
             r = self.webhost.request('GET', 'unhandled_error', no_prefix=True)
             self.assertEqual(r.status_code, 500)
+            # The worker forwards the traceback to the host asynchronously, so
+            # wait for it before check_log_unhandled_error reads its snapshot.
+            self.wait_for_host_log('ZeroDivisionError: division by zero')
             # https://github.com/Azure/azure-functions-host/issues/2706
             # self.assertIn('ZeroDivisionError', r.text)
 
diff --git a/workers/tests/unittests/test_vendored_protobuf.py b/workers/tests/unittests/test_vendored_protobuf.py
index 09ebaf9be..37655f0db 100644
--- a/workers/tests/unittests/test_vendored_protobuf.py
+++ b/workers/tests/unittests/test_vendored_protobuf.py
@@ -443,6 +443,86 @@ def test_launcher_override_can_force_no_activation(self):
         )
         self.assertIn("OK", result.stdout)
 
+    def _write_app_protobuf(self, root: Path, version: str) -> Path:
+        """Create a fake function app ``google.protobuf`` with a given
+        ``__version__`` under a simulated ``.python_packages`` tree.
+        Returns the script-root path (parent of ``.python_packages``).
+        """
+        site_packages = (
+            root / ".python_packages" / "lib" / "site-packages"
+        )
+        pkg = site_packages / "google" / "protobuf"
+        pkg.mkdir(parents=True)
+        (site_packages / "google" / "__init__.py").write_text(
+            "# Stub function app google namespace package.\n",
+            encoding="utf-8",
+        )
+        (pkg / "__init__.py").write_text(
+            f'__version__ = "{version}"\n', encoding="utf-8",
+        )
+        return root
+
+    @unittest.skipUnless(
+        _vendored_protobuf_present(),
+        "Vendored protobuf is not populated. Run "
+        "`python eng/scripts/vendor_deps.py --target "
+        "workers/azure_functions_worker/_vendored` first.",
+    )
+    def test_newer_app_protobuf_does_not_activate_vendored(self):
+        """A function app protobuf newer than the vendored copy (e.g.
+        protobuf 6.x from an extension) must NOT activate the vendored
+        fallback. The worker's stubs run on the newer runtime, and forcing
+        the newer copy onto the older vendored one would raise a
+        ``VersionError``.
+        """
+        newer_root = self._write_app_protobuf(
+            Path(self.tmp_dir), "6.99.0")
+        newer_site = (
+            newer_root / ".python_packages" / "lib" / "site-packages"
+        )
+
+        code = textwrap.dedent(
+            """
+            import os
+            import sys
+            import azure_functions_worker  # noqa: F401  triggers detection
+
+            # A newer function app protobuf must NOT force pure-Python.
+            assert (
+                os.environ.get("PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION")
+                is None
+            ), (
+                "worker forced pure-Python despite the app protobuf "
+                "being newer than the vendored copy"
+            )
+
+            # The vendored copy must NOT be aliased over google.protobuf.
+            top_pb = sys.modules.get("google.protobuf")
+            if top_pb is not None:
+                assert "_vendored" not in (top_pb.__file__ or ""), (
+                    "vendored protobuf was activated despite the app "
+                    "protobuf being newer"
+                )
+            print("OK")
+            """
+        )
+
+        result = self._run_subprocess(
+            code,
+            extra_path=newer_site,
+            script_root=str(newer_root),
+        )
+        self.assertEqual(
+            result.returncode,
+            0,
+            msg=(
+                "Worker activated the vendored fallback for a newer "
+                "app protobuf.\n"
+                f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
+            ),
+        )
+        self.assertIn("OK", result.stdout)
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/workers/tests/utils/constants.py b/workers/tests/utils/constants.py
index 0a2ca0b52..dfd791264 100644
--- a/workers/tests/utils/constants.py
+++ b/workers/tests/utils/constants.py
@@ -81,6 +81,10 @@
 CONSUMPTION_DOCKER_TEST = "CONSUMPTION_DOCKER_TEST"
 DEDICATED_DOCKER_TEST = "DEDICATED_DOCKER_TEST"
 
+# Master key used in the webhost Secrets/host.json template; required to call
+# the host's protected /admin endpoints (e.g. /admin/host/status).
+MASTER_KEY = "testMasterKey"
+
 # Paths anchored on this file's location rather than on PROJECT_ROOT/TESTS_ROOT.
 # The `tests.utils` package exists in multiple trees in this repo
 # (workers/tests/utils and runtimes/v1/tests/utils), so PROJECT_ROOT/TESTS_ROOT
diff --git a/workers/tests/utils/testutils.py b/workers/tests/utils/testutils.py
index f90bd3258..4a9dda6d3 100644
--- a/workers/tests/utils/testutils.py
+++ b/workers/tests/utils/testutils.py
@@ -38,6 +38,7 @@
     CONSUMPTION_DOCKER_TEST,
     DEDICATED_DOCKER_TEST,
     EXTENSIONS_CSPROJ_TEMPLATE,
+    MASTER_KEY,
     PROJECT_ROOT,
     PYAZURE_INTEGRATION_TEST,
     PYAZURE_WEBHOST_DEBUG,
@@ -322,6 +323,36 @@ def _run_test(self, test, *args, **kwargs):
                     if test_exception is not None:
                         raise test_exception
 
+    def wait_for_host_log(self, substring: str,
+                          timeout: float = 10.0,
+                          poll_interval: float = 0.5) -> bool:
+        """Wait until `substring` appears in the host stdout written so far.
+
+        The worker forwards exception logs to the host over gRPC, so a log
+        line can land in the host's stdout slightly after the corresponding
+        HTTP response is returned. The check_log_* assertions read a snapshot
+        of host_out taken right after the test method returns; without waiting,
+        that snapshot can miss the late-arriving line, making such tests flaky.
+        Call this at the end of a test_* method so the snapshot includes the
+        line. Best-effort: returns True if found, False on timeout (the
+        check_log_* assertion still runs and decides).
+        """
+        if self.host_stdout is None:
+            return True
+        start = self.host_stdout.tell()
+        deadline = time.time() + timeout
+        try:
+            while time.time() < deadline:
+                self.host_stdout.seek(start)
+                if substring in self.host_stdout.read():
+                    return True
+                time.sleep(poll_interval)
+            return False
+        finally:
+            # Restore the read position so the framework captures host_out
+            # from the same point regardless of our polling reads.
+            self.host_stdout.seek(start)
+
 
 # This is not supported in 3.13+
 if sys.version_info.minor < 13:
@@ -806,6 +837,75 @@ def is_healthy(self):
         r = self.request('GET', '', no_prefix=True)
         return 200 <= r.status_code < 300
 
+    def wait_until_ready(self, timeout: float = 60.0,
+                         poll_interval: float = 0.5) -> bool:
+        """Poll the host until it is running AND has registered functions.
+
+        Readiness is confirmed in two phases:
+
+        1. ``/admin/host/status`` reports ``{"state": "Running", ...}``,
+           which means the host has started and the worker has connected.
+        2. ``/admin/functions`` returns a non-empty list, which means the
+           worker finished importing ``function_app.py`` and the host has
+           registered the routes.
+
+        Phase 2 matters for the v2 programming model with a heavy
+        ``function_app.py`` (e.g. the ServiceBus SDK app, which imports
+        uamqp): the host can report ``Running`` while the worker is still
+        indexing, so HTTP routes intermittently return 404. Waiting for the
+        function list to be populated closes that gap.
+
+        Both admin endpoints are protected by the master key, so the
+        requests must include it; otherwise the host replies 401 and we
+        would block until the full timeout on every webhost start. If the
+        host does not expose ``/admin/functions`` (404), we fall back to
+        treating ``Running`` as ready so older hosts are not regressed.
+        """
+        deadline = time.time() + timeout
+        status_url = self._addr + '/admin/host/status'
+        functions_url = self._addr + '/admin/functions'
+        headers = {'x-functions-key': MASTER_KEY}
+        last_state = None
+        running = False
+        while time.time() < deadline:
+            if self._proc.poll() is not None:
+                # Host process exited.
+                return False
+            try:
+                if not running:
+                    r = requests.get(status_url, headers=headers, timeout=5)
+                    if r.status_code == 200:
+                        try:
+                            last_state = r.json().get('state')
+                        except ValueError:
+                            last_state = None
+                        running = last_state == 'Running'
+
+                if running:
+                    fr = requests.get(functions_url, headers=headers,
+                                      timeout=5)
+                    if fr.status_code == 404:
+                        # Functions admin endpoint not available on this
+                        # host; Running is the best signal we have.
+                        return True
+                    if fr.status_code == 200:
+                        try:
+                            functions = fr.json()
+                        except ValueError:
+                            functions = None
+                        if functions:
+                            return True
+            except requests.RequestException:
+                pass
+            time.sleep(poll_interval)
+        logging.getLogger('webhosttests').warning(
+            "Webhost did not become ready within %.0fs "
+            "(host running: %s, last state: %r, functions registered: no). "
+            "The function app likely failed to index; see the captured "
+            "WebHost log for the worker error.",
+            timeout, running, last_state)
+        return False
+
     def request(self, meth, funcname, *args, **kwargs):
         request_method = getattr(requests, meth.lower())
         params = dict(kwargs.pop('params', {}))
@@ -980,10 +1080,15 @@ def start_webhost(*, script_dir=None, stdout=None):
 
     proc = popen_webhost(stdout=stdout, stderr=subprocess.STDOUT,
                          script_root=script_root, port=port)
-    time.sleep(10)  # Giving host some time to start fully.
 
     addr = f'http://{LOCALHOST}:{port}'
-    return _WebHostProxy(proc, addr)
+    proxy = _WebHostProxy(proc, addr)
+    # Poll the host's /admin/host/status until the host reports `Running`,
+    # rather than relying on a fixed sleep. The previous `time.sleep(10)`
+    # was racy on slower agents (Python 3.9-3.11 cold starts in particular)
+    # which caused intermittent test failures like flaky `test_unhandled_error`.
+    proxy.wait_until_ready(timeout=60.0)
+    return proxy
 
 
 def create_dummy_dispatcher():