From 2e6861f9990b1b50ab9ff7c20621474e3aae5fa1 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 05:59:55 -0400
Subject: [PATCH 01/10] feat(api-core): add ClientInterceptor and
 apply_interceptors helper

---
 .../google/api_core/grpc_helpers.py           | 35 ++++++-
 .../tests/unit/test_grpc_helpers.py           | 94 ++++++++++++++++++-
 2 files changed, 125 insertions(+), 4 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 263079e7d1f7..ab944b240198 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -17,7 +17,7 @@
 import collections
 import functools
 import warnings
-from typing import Generic, Iterator, Optional, TypeVar
+from typing import Generic, Iterator, Optional, Sequence, TypeVar, Union
 
 import google.auth
 import google.auth.credentials
@@ -25,7 +25,6 @@
 import google.auth.transport.requests
 import google.protobuf
 import grpc
-
 from google.api_core import exceptions, general_helpers
 
 # The list of gRPC Callable interfaces that return iterators.
@@ -34,6 +33,14 @@
 # denotes the proto response type for grpc calls
 P = TypeVar("P")
 
+# Type alias representing any client-side gRPC interceptor
+ClientInterceptor = Union[
+    grpc.UnaryUnaryClientInterceptor,
+    grpc.UnaryStreamClientInterceptor,
+    grpc.StreamUnaryClientInterceptor,
+    grpc.StreamStreamClientInterceptor,
+]
+
 
 def _patch_callable_name(callable_):
     """Fix-up gRPC callable attributes.
@@ -419,6 +426,30 @@ def _modify_target_for_direct_path(target: str) -> str:
     return target
 
 
+def apply_interceptors(
+    channel: grpc.Channel,
+    interceptors: Optional[Sequence[ClientInterceptor]] = None,
+) -> grpc.Channel:
+    """Applies a sequence of interceptors to a gRPC channel.
+
+    The interceptors are applied in the order provided, wrapping the channel
+    sequentially.
+
+    Args:
+        channel (grpc.Channel): The channel to intercept.
+        interceptors (Optional[Sequence[ClientInterceptor]]): An optional sequence
+            of client interceptors to apply.
+
+    Returns:
+        grpc.Channel: The intercepted channel, or the original channel if no
+            interceptors were provided.
+    """
+    if interceptors:
+        for interceptor in interceptors:
+            channel = grpc.intercept_channel(channel, interceptor)
+    return channel
+
+
 _MethodCall = collections.namedtuple(
     "_MethodCall", ("request", "timeout", "metadata", "credentials", "compression")
 )
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 69281d58109b..677fc15ce2d3 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -24,9 +24,8 @@
     pytest.skip("No GRPC", allow_module_level=True)
 
 import google.auth.credentials
-from google.longrunning import operations_pb2
-
 from google.api_core import exceptions, grpc_helpers
+from google.longrunning import operations_pb2
 
 
 def test__patch_callable_name():
@@ -932,3 +931,94 @@ def test_subscribe_unsubscribe(self):
     def test_close(self):
         channel = grpc_helpers.ChannelStub()
         assert channel.close() is None
+
+
+@pytest.mark.parametrize("falsy_interceptors", [None, [], ()])
+def test_apply_interceptors_passthrough(falsy_interceptors):
+    """Verify that falsy or empty interceptor sequences return the channel unmodified."""
+    mock_channel = mock.Mock()
+    result = grpc_helpers.apply_interceptors(mock_channel, falsy_interceptors)
+    assert result is mock_channel
+
+
+@pytest.mark.parametrize("count", [1, 2, 3])
+def test_apply_interceptors_wrapping(count):
+    """Verify that interceptors are wrapped sequentially in the order provided.
+
+    When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors
+    must pass the base channel and i_0 to grpc.intercept_channel, then pass the
+    resulting wrapped channel and i_1 to grpc.intercept_channel, and so on.
+    This ensures each subsequent interceptor wraps the preceding channel state.
+    """
+    mock_channel = mock.Mock(name="base_channel")
+    # Generate distinct mock interceptors and the expected wrapped channel returns for each step
+    interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
+    wrapped_channels = [mock.Mock(name=f"wrapped_channel_{i}") for i in range(count)]
+
+    with mock.patch(
+        "grpc.intercept_channel", side_effect=wrapped_channels
+    ) as mock_intercept:
+        result = grpc_helpers.apply_interceptors(mock_channel, interceptors)
+
+    # The final return value must be the outermost wrapped channel from the final loop iteration
+    assert result is wrapped_channels[-1]
+    assert mock_intercept.call_count == count
+
+    # Construct the expected sequential chaining: (base, i_0) -> (wrapped_0, i_1) -> ...
+    expected_calls = []
+    current_channel = mock_channel
+    for i, interceptor in enumerate(interceptors):
+        expected_calls.append(mock.call(current_channel, interceptor))
+        current_channel = wrapped_channels[i]
+
+    mock_intercept.assert_has_calls(expected_calls)
+
+
+def test_apply_interceptors_execution_order():
+    """Verify runtime execution order (onion model) when invoking an RPC on an intercepted channel.
+
+    In gRPC Python, sequential wrapping via grpc.intercept_channel(channel, i) creates an
+    'onion' layer where the LAST applied interceptor becomes the OUTSIDE layer.
+    Therefore, given [i1, i2]:
+      - i1 wraps the raw channel (innermost layer)
+      - i2 wraps the result of (channel + i1) (outermost layer)
+
+    During an RPC invocation:
+      1. i2 intercepts the call first (request inbound / pre-call)
+      2. i2 calls continuation(), which triggers i1
+      3. i1 calls continuation(), which reaches the channel stub / network
+      4. i1 post-call logic finishes
+      5. i2 post-call logic finishes
+    """
+    execution_order = []
+
+    class OrderInterceptor(grpc.UnaryUnaryClientInterceptor):
+        def __init__(self, name):
+            self.name = name
+
+        def intercept_unary_unary(self, continuation, client_call_details, request):
+            execution_order.append(f"{self.name}_start")
+            response = continuation(client_call_details, request)
+            execution_order.append(f"{self.name}_end")
+            return response
+
+    i1 = OrderInterceptor("i1")
+    i2 = OrderInterceptor("i2")
+
+    mock_channel = mock.Mock(spec=grpc.Channel)
+    mock_callable = mock.Mock(spec=grpc.UnaryUnaryMultiCallable)
+    mock_call = mock.Mock(spec=grpc.Call)
+    expected_response = operations_pb2.Operation(name="test_op")
+    mock_callable.with_call.return_value = (expected_response, mock_call)
+    mock_channel.unary_unary.return_value = mock_callable
+
+    # Apply interceptors in sequence [i1, i2]
+    intercepted_channel = grpc_helpers.apply_interceptors(mock_channel, [i1, i2])
+
+    # Trigger a unary RPC through the intercepted channel
+    stub = operations_pb2.OperationsStub(intercepted_channel)
+    response = stub.GetOperation(operations_pb2.GetOperationRequest(name="test_op"))
+
+    assert response.name == "test_op"
+    # Verify i2 executed as the outer layer surrounding i1
+    assert execution_order == ["i2_start", "i1_start", "i1_end", "i2_end"]

From 84e4cd55babd5de780aaf61b978963ab9ac1d4e3 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 06:30:02 -0400
Subject: [PATCH 02/10] fix(api-core): unpack interceptors directly into
 grpc.intercept_channel

---
 .../google/api_core/grpc_helpers.py           |  7 ++-
 .../tests/unit/test_grpc_helpers.py           | 49 +++++++------------
 2 files changed, 22 insertions(+), 34 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index ab944b240198..bb2a3523d735 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -432,8 +432,8 @@ def apply_interceptors(
 ) -> grpc.Channel:
     """Applies a sequence of interceptors to a gRPC channel.
 
-    The interceptors are applied in the order provided, wrapping the channel
-    sequentially.
+    The interceptors are applied in the order provided, such that the first
+    interceptor in the sequence is the outermost layer (executes first).
 
     Args:
         channel (grpc.Channel): The channel to intercept.
@@ -445,8 +445,7 @@ def apply_interceptors(
             interceptors were provided.
     """
     if interceptors:
-        for interceptor in interceptors:
-            channel = grpc.intercept_channel(channel, interceptor)
+        return grpc.intercept_channel(channel, *interceptors)
     return channel
 
 
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 677fc15ce2d3..b9fc553e6555 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -943,52 +943,41 @@ def test_apply_interceptors_passthrough(falsy_interceptors):
 
 @pytest.mark.parametrize("count", [1, 2, 3])
 def test_apply_interceptors_wrapping(count):
-    """Verify that interceptors are wrapped sequentially in the order provided.
+    """Verify that interceptors are passed to grpc.intercept_channel in a single call.
 
     When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors
-    must pass the base channel and i_0 to grpc.intercept_channel, then pass the
-    resulting wrapped channel and i_1 to grpc.intercept_channel, and so on.
-    This ensures each subsequent interceptor wraps the preceding channel state.
+    must pass the base channel and all interceptors unpacked (*interceptors) to
+    grpc.intercept_channel. This creates a single intercepted channel wrapper rather than
+    multiple nested wrappers.
     """
     mock_channel = mock.Mock(name="base_channel")
-    # Generate distinct mock interceptors and the expected wrapped channel returns for each step
+    mock_intercepted = mock.Mock(name="intercepted_channel")
     interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
-    wrapped_channels = [mock.Mock(name=f"wrapped_channel_{i}") for i in range(count)]
 
     with mock.patch(
-        "grpc.intercept_channel", side_effect=wrapped_channels
+        "grpc.intercept_channel", return_value=mock_intercepted
     ) as mock_intercept:
         result = grpc_helpers.apply_interceptors(mock_channel, interceptors)
 
-    # The final return value must be the outermost wrapped channel from the final loop iteration
-    assert result is wrapped_channels[-1]
-    assert mock_intercept.call_count == count
-
-    # Construct the expected sequential chaining: (base, i_0) -> (wrapped_0, i_1) -> ...
-    expected_calls = []
-    current_channel = mock_channel
-    for i, interceptor in enumerate(interceptors):
-        expected_calls.append(mock.call(current_channel, interceptor))
-        current_channel = wrapped_channels[i]
-
-    mock_intercept.assert_has_calls(expected_calls)
+    assert result is mock_intercepted
+    mock_intercept.assert_called_once_with(mock_channel, *interceptors)
 
 
 def test_apply_interceptors_execution_order():
     """Verify runtime execution order (onion model) when invoking an RPC on an intercepted channel.
 
-    In gRPC Python, sequential wrapping via grpc.intercept_channel(channel, i) creates an
-    'onion' layer where the LAST applied interceptor becomes the OUTSIDE layer.
+    In standard gRPC Python, grpc.intercept_channel(channel, *interceptors) processes
+    the interceptor list such that the first interceptor in the sequence is the outermost layer.
     Therefore, given [i1, i2]:
-      - i1 wraps the raw channel (innermost layer)
-      - i2 wraps the result of (channel + i1) (outermost layer)
+      - i1 is the outermost layer (executes first on outbound request)
+      - i2 is the inner layer (executes second on outbound request)
 
     During an RPC invocation:
-      1. i2 intercepts the call first (request inbound / pre-call)
-      2. i2 calls continuation(), which triggers i1
-      3. i1 calls continuation(), which reaches the channel stub / network
-      4. i1 post-call logic finishes
-      5. i2 post-call logic finishes
+      1. i1 intercepts the call first (request inbound / pre-call)
+      2. i1 calls continuation(), which triggers i2
+      3. i2 calls continuation(), which reaches the channel stub / network
+      4. i2 post-call logic finishes
+      5. i1 post-call logic finishes
     """
     execution_order = []
 
@@ -1020,5 +1009,5 @@ def intercept_unary_unary(self, continuation, client_call_details, request):
     response = stub.GetOperation(operations_pb2.GetOperationRequest(name="test_op"))
 
     assert response.name == "test_op"
-    # Verify i2 executed as the outer layer surrounding i1
-    assert execution_order == ["i2_start", "i1_start", "i1_end", "i2_end"]
+    # Verify i1 executed as the outer layer surrounding i2
+    assert execution_order == ["i1_start", "i2_start", "i2_end", "i1_end"]

From 9ea6cc6339debf68c268ceaaddec711512a677f9 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 09:26:26 -0400
Subject: [PATCH 03/10] docs(api-core): clarify apply_interceptors execution
 order in docstring

---
 packages/google-api-core/google/api_core/grpc_helpers.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index bb2a3523d735..2f0ef9631dd8 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -430,10 +430,10 @@ def apply_interceptors(
     channel: grpc.Channel,
     interceptors: Optional[Sequence[ClientInterceptor]] = None,
 ) -> grpc.Channel:
-    """Applies a sequence of interceptors to a gRPC channel.
+    """Applies client interceptors to a gRPC channel.
 
-    The interceptors are applied in the order provided, such that the first
-    interceptor in the sequence is the outermost layer (executes first).
+    The first interceptor in the sequence is the outermost layer: it
+    executes first on outbound requests and last on inbound responses.
 
     Args:
         channel (grpc.Channel): The channel to intercept.

From ea38a9866b087bbd9ba1de3f20fdebe2707f321c Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 09:45:16 -0400
Subject: [PATCH 04/10] test(api-core): remove redundant execution order test
 and simplify interceptor unit tests

---
 .../tests/unit/test_grpc_helpers.py           | 55 +------------------
 1 file changed, 2 insertions(+), 53 deletions(-)

diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index b9fc553e6555..0dad58217545 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -943,12 +943,11 @@ def test_apply_interceptors_passthrough(falsy_interceptors):
 
 @pytest.mark.parametrize("count", [1, 2, 3])
 def test_apply_interceptors_wrapping(count):
-    """Verify that interceptors are passed to grpc.intercept_channel in a single call.
+    """Verify that interceptors are passed to grpc.intercept_channel unpacked in a single call.
 
     When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors
     must pass the base channel and all interceptors unpacked (*interceptors) to
-    grpc.intercept_channel. This creates a single intercepted channel wrapper rather than
-    multiple nested wrappers.
+    grpc.intercept_channel.
     """
     mock_channel = mock.Mock(name="base_channel")
     mock_intercepted = mock.Mock(name="intercepted_channel")
@@ -961,53 +960,3 @@ def test_apply_interceptors_wrapping(count):
 
     assert result is mock_intercepted
     mock_intercept.assert_called_once_with(mock_channel, *interceptors)
-
-
-def test_apply_interceptors_execution_order():
-    """Verify runtime execution order (onion model) when invoking an RPC on an intercepted channel.
-
-    In standard gRPC Python, grpc.intercept_channel(channel, *interceptors) processes
-    the interceptor list such that the first interceptor in the sequence is the outermost layer.
-    Therefore, given [i1, i2]:
-      - i1 is the outermost layer (executes first on outbound request)
-      - i2 is the inner layer (executes second on outbound request)
-
-    During an RPC invocation:
-      1. i1 intercepts the call first (request inbound / pre-call)
-      2. i1 calls continuation(), which triggers i2
-      3. i2 calls continuation(), which reaches the channel stub / network
-      4. i2 post-call logic finishes
-      5. i1 post-call logic finishes
-    """
-    execution_order = []
-
-    class OrderInterceptor(grpc.UnaryUnaryClientInterceptor):
-        def __init__(self, name):
-            self.name = name
-
-        def intercept_unary_unary(self, continuation, client_call_details, request):
-            execution_order.append(f"{self.name}_start")
-            response = continuation(client_call_details, request)
-            execution_order.append(f"{self.name}_end")
-            return response
-
-    i1 = OrderInterceptor("i1")
-    i2 = OrderInterceptor("i2")
-
-    mock_channel = mock.Mock(spec=grpc.Channel)
-    mock_callable = mock.Mock(spec=grpc.UnaryUnaryMultiCallable)
-    mock_call = mock.Mock(spec=grpc.Call)
-    expected_response = operations_pb2.Operation(name="test_op")
-    mock_callable.with_call.return_value = (expected_response, mock_call)
-    mock_channel.unary_unary.return_value = mock_callable
-
-    # Apply interceptors in sequence [i1, i2]
-    intercepted_channel = grpc_helpers.apply_interceptors(mock_channel, [i1, i2])
-
-    # Trigger a unary RPC through the intercepted channel
-    stub = operations_pb2.OperationsStub(intercepted_channel)
-    response = stub.GetOperation(operations_pb2.GetOperationRequest(name="test_op"))
-
-    assert response.name == "test_op"
-    # Verify i1 executed as the outer layer surrounding i2
-    assert execution_order == ["i1_start", "i2_start", "i2_end", "i1_end"]

From 8e65db065a23465d500f59d00aa7efa3246f18e1 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 10:08:49 -0400
Subject: [PATCH 05/10] test(api-core): align mock variable naming to Option 1
 convention

---
 .../tests/unit/test_grpc_helpers.py           | 24 ++++++++++---------
 1 file changed, 13 insertions(+), 11 deletions(-)

diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 0dad58217545..9e41bab853df 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -936,9 +936,9 @@ def test_close(self):
 @pytest.mark.parametrize("falsy_interceptors", [None, [], ()])
 def test_apply_interceptors_passthrough(falsy_interceptors):
     """Verify that falsy or empty interceptor sequences return the channel unmodified."""
-    mock_channel = mock.Mock()
-    result = grpc_helpers.apply_interceptors(mock_channel, falsy_interceptors)
-    assert result is mock_channel
+    mock_base_channel = mock.Mock(name="base_channel")
+    result = grpc_helpers.apply_interceptors(mock_base_channel, falsy_interceptors)
+    assert result is mock_base_channel
 
 
 @pytest.mark.parametrize("count", [1, 2, 3])
@@ -949,14 +949,16 @@ def test_apply_interceptors_wrapping(count):
     must pass the base channel and all interceptors unpacked (*interceptors) to
     grpc.intercept_channel.
     """
-    mock_channel = mock.Mock(name="base_channel")
-    mock_intercepted = mock.Mock(name="intercepted_channel")
-    interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
+    mock_base_channel = mock.Mock(name="base_channel")
+    mock_wrapped_channel = mock.Mock(name="wrapped_channel")
+    mock_interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
 
     with mock.patch(
-        "grpc.intercept_channel", return_value=mock_intercepted
-    ) as mock_intercept:
-        result = grpc_helpers.apply_interceptors(mock_channel, interceptors)
+        "grpc.intercept_channel", return_value=mock_wrapped_channel
+    ) as mock_intercept_channel:
+        result = grpc_helpers.apply_interceptors(mock_base_channel, mock_interceptors)
 
-    assert result is mock_intercepted
-    mock_intercept.assert_called_once_with(mock_channel, *interceptors)
+    assert result is mock_wrapped_channel
+    mock_intercept_channel.assert_called_once_with(
+        mock_base_channel, *mock_interceptors
+    )

From b41b0c1cad30d94609cae0ad095f4b9ffb93b993 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Fri, 28 Aug 2026 09:51:22 -0400
Subject: [PATCH 06/10] feat(api-core): update default env var to
 GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED

- Set is_otel_capabilities_enabled default env_var to GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED

- Activate fail-fast FeatureGatingError experimental path when tracer_provider is set without env var

- Update unit tests to verify experimental gating behavior
---
 .../google/api_core/_observability.py         |  2 +-
 .../tests/unit/test_observability.py          | 43 +++++++++++++++++--
 2 files changed, 41 insertions(+), 4 deletions(-)

diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py
index a36df8b39599..b1b71b056658 100644
--- a/packages/google-api-core/google/api_core/_observability.py
+++ b/packages/google-api-core/google/api_core/_observability.py
@@ -26,7 +26,7 @@
 
 def is_otel_capabilities_enabled(
     client_options: Optional[ClientOptions | dict[str, Any]] = None,
-    env_var: str = "GOOGLE_CLOUD_PYTHON_TRACING_ENABLED",
+    env_var: str = "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
 ) -> bool:
     """Checks if OTel capabilities are enabled and installed.
 
diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py
index fc63023aadcd..d39edb806040 100644
--- a/packages/google-api-core/tests/unit/test_observability.py
+++ b/packages/google-api-core/tests/unit/test_observability.py
@@ -15,17 +15,19 @@
 import sys
 from unittest import mock
 
+import pytest
 from google.api_core import _observability
+from google.api_core._feature_gating_helpers import FeatureGatingError
 from google.api_core.client_options import ClientOptions
 
 
 def test_is_otel_capabilities_enabled_disabled(monkeypatch):
-    monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "false")
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
     assert not _observability.is_otel_capabilities_enabled()
 
 
 def test_is_otel_capabilities_enabled_otel_missing(monkeypatch):
-    monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "true")
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
     # Simulate OTel not being installed by blocking imports
     monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", None)
 
@@ -33,7 +35,7 @@ def test_is_otel_capabilities_enabled_otel_missing(monkeypatch):
 
 
 def test_is_otel_capabilities_enabled_otel_installed(monkeypatch):
-    monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "true")
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
 
     mock_otel = mock.Mock()
     mock_otel_grpc = mock_otel.instrumentation.grpc
@@ -49,6 +51,41 @@ def test_is_otel_capabilities_enabled_otel_installed(monkeypatch):
     assert _observability.is_otel_capabilities_enabled()
 
 
+def test_is_otel_capabilities_enabled_experimental_requires_env_var(monkeypatch):
+    """Proves that passing client_options with tracer_provider without the experimental
+    env var set to 'true' raises FeatureGatingError (Fail Fast).
+    """
+    monkeypatch.delenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", raising=False)
+    options = ClientOptions(tracer_provider=object())
+
+    with pytest.raises(
+        FeatureGatingError,
+        match="requires GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
+    ):
+        _observability.is_otel_capabilities_enabled(options)
+
+
+def test_is_otel_capabilities_enabled_experimental_enabled_with_config(monkeypatch):
+    """Proves that when GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED=true and tracer_provider
+    is supplied via client_options, is_otel_capabilities_enabled returns True.
+    """
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
+
+    mock_otel = mock.Mock()
+    mock_otel_grpc = mock_otel.instrumentation.grpc
+
+    monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation
+    )
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
+    )
+
+    options = ClientOptions(tracer_provider=object())
+    assert _observability.is_otel_capabilities_enabled(options)
+
+
 def test_apply_otel_capabilities_to_channel_enabled_otel_installed(monkeypatch):
     mock_channel = mock.Mock()
     mock_intercepted_channel = mock.Mock()

From 8b2dc1fc458d505bf6bfc5ac1032d24a14f51018 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Mon, 31 Aug 2026 08:15:09 -0400
Subject: [PATCH 07/10] feat(api-core): add ChannelWrapper and
 apply_channel_wrappers helper

---
 .../google/api_core/grpc_helpers.py           |  63 ++++++++---
 .../tests/unit/test_grpc_helpers.py           | 106 ++++++++++++++----
 2 files changed, 136 insertions(+), 33 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 2f0ef9631dd8..5f9baf4cf12f 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -17,7 +17,16 @@
 import collections
 import functools
 import warnings
-from typing import Generic, Iterator, Optional, Sequence, TypeVar, Union
+from typing import (
+    Callable,
+    Generic,
+    Iterator,
+    Optional,
+    Sequence,
+    TypeVar,
+    Union,
+    get_args,
+)
 
 import google.auth
 import google.auth.credentials
@@ -41,6 +50,15 @@
     grpc.StreamStreamClientInterceptor,
 ]
 
+# Runtime tuple of gRPC client interceptor base classes for isinstance checks
+_CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
+
+# Type alias representing a channel-wrapping callable
+ChannelWrapperCallable = Callable[[grpc.Channel], grpc.Channel]
+
+# Generic type alias representing any channel wrapper (interceptor or callable)
+ChannelWrapper = Union[ClientInterceptor, ChannelWrapperCallable]
+
 
 def _patch_callable_name(callable_):
     """Fix-up gRPC callable attributes.
@@ -426,27 +444,44 @@ def _modify_target_for_direct_path(target: str) -> str:
     return target
 
 
-def apply_interceptors(
+def apply_channel_wrappers(
     channel: grpc.Channel,
-    interceptors: Optional[Sequence[ClientInterceptor]] = None,
+    wrappers: Optional[Sequence[ChannelWrapper]] = None,
 ) -> grpc.Channel:
-    """Applies client interceptors to a gRPC channel.
+    """Applies channel wrappers (client interceptors or channel-wrapping callables) to a gRPC channel.
 
-    The first interceptor in the sequence is the outermost layer: it
-    executes first on outbound requests and last on inbound responses.
+    Executes in reverse order so the first wrapper in the sequence becomes the
+    outermost layer on outbound requests and the innermost layer on inbound responses.
 
     Args:
-        channel (grpc.Channel): The channel to intercept.
-        interceptors (Optional[Sequence[ClientInterceptor]]): An optional sequence
-            of client interceptors to apply.
+        channel (grpc.Channel): The channel to wrap.
+        wrappers (Optional[Sequence[ChannelWrapper]]):
+            An optional sequence of client interceptors or channel-wrapping
+            callables to apply.
 
     Returns:
-        grpc.Channel: The intercepted channel, or the original channel if no
-            interceptors were provided.
+        grpc.Channel: The wrapped channel, or the original channel if no
+            wrappers were provided.
+
+    Raises:
+        TypeError: If an item in ``wrappers`` is neither a gRPC ClientInterceptor
+            nor a Callable[[Channel], Channel].
     """
-    if interceptors:
-        return grpc.intercept_channel(channel, *interceptors)
-    return channel
+    if not wrappers:
+        return channel
+
+    modified_channel = channel
+    for wrapper in reversed(list(wrappers)):
+        if isinstance(wrapper, _CLIENT_INTERCEPTOR_CLASSES):
+            modified_channel = grpc.intercept_channel(modified_channel, wrapper)
+        elif callable(wrapper):
+            modified_channel = wrapper(modified_channel)
+        else:
+            raise TypeError(
+                f"Expected ChannelWrapper (ClientInterceptor or Callable[[Channel], Channel]), got {type(wrapper).__name__}"
+            )
+
+    return modified_channel
 
 
 _MethodCall = collections.namedtuple(
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 9e41bab853df..8d91e63f15f7 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -933,32 +933,100 @@ def test_close(self):
         assert channel.close() is None
 
 
-@pytest.mark.parametrize("falsy_interceptors", [None, [], ()])
-def test_apply_interceptors_passthrough(falsy_interceptors):
-    """Verify that falsy or empty interceptor sequences return the channel unmodified."""
+@pytest.mark.parametrize("falsy_wrappers", [None, [], ()])
+def test_apply_channel_wrappers_passthrough(falsy_wrappers):
+    """Verify that falsy or empty wrapper sequences return the channel unmodified."""
     mock_base_channel = mock.Mock(name="base_channel")
-    result = grpc_helpers.apply_interceptors(mock_base_channel, falsy_interceptors)
+    result = grpc_helpers.apply_channel_wrappers(mock_base_channel, falsy_wrappers)
     assert result is mock_base_channel
 
 
-@pytest.mark.parametrize("count", [1, 2, 3])
-def test_apply_interceptors_wrapping(count):
-    """Verify that interceptors are passed to grpc.intercept_channel unpacked in a single call.
+def test_apply_channel_wrappers_grpc_client_interceptors():
+    """Verify that standard gRPC ClientInterceptor instances are applied via grpc.intercept_channel."""
+
+    class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor):
+        def intercept_unary_unary(self, continuation, client_call_details, request):
+            return continuation(client_call_details, request)
+
+    class DummyStreamInterceptor(grpc.StreamStreamClientInterceptor):
+        def intercept_stream_stream(
+            self, continuation, client_call_details, request_iterator
+        ):
+            return continuation(client_call_details, request_iterator)
+
+    interceptor1 = DummyUnaryInterceptor()
+    interceptor2 = DummyStreamInterceptor()
 
-    When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors
-    must pass the base channel and all interceptors unpacked (*interceptors) to
-    grpc.intercept_channel.
-    """
     mock_base_channel = mock.Mock(name="base_channel")
-    mock_wrapped_channel = mock.Mock(name="wrapped_channel")
-    mock_interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
+    mock_chan_after_i2 = mock.Mock(name="chan_after_i2")
+    mock_chan_after_i1 = mock.Mock(name="chan_after_i1")
 
     with mock.patch(
-        "grpc.intercept_channel", return_value=mock_wrapped_channel
-    ) as mock_intercept_channel:
-        result = grpc_helpers.apply_interceptors(mock_base_channel, mock_interceptors)
+        "grpc.intercept_channel",
+        side_effect=[mock_chan_after_i2, mock_chan_after_i1],
+    ) as mock_intercept:
+        result = grpc_helpers.apply_channel_wrappers(
+            mock_base_channel, [interceptor1, interceptor2]
+        )
 
-    assert result is mock_wrapped_channel
-    mock_intercept_channel.assert_called_once_with(
-        mock_base_channel, *mock_interceptors
+    assert result is mock_chan_after_i1
+    assert mock_intercept.call_count == 2
+    # Executed in reverse order so interceptor1 is outermost
+    mock_intercept.assert_has_calls(
+        [
+            mock.call(mock_base_channel, interceptor2),
+            mock.call(mock_chan_after_i2, interceptor1),
+        ]
     )
+
+
+def test_apply_channel_wrappers_callables():
+    """Verify that channel-wrapping callables Callable[[Channel], Channel] are invoked in sequence."""
+    mock_base_channel = mock.Mock(name="base_channel")
+    mock_chan_1 = mock.Mock(name="chan_1")
+    mock_chan_2 = mock.Mock(name="chan_2")
+
+    wrapper1 = mock.Mock(side_effect=lambda ch: mock_chan_2)
+    wrapper2 = mock.Mock(side_effect=lambda ch: mock_chan_1)
+
+    result = grpc_helpers.apply_channel_wrappers(
+        mock_base_channel, [wrapper1, wrapper2]
+    )
+
+    assert result is mock_chan_2
+    # Executed in reverse order: wrapper2 runs first on base channel, then wrapper1
+    wrapper2.assert_called_once_with(mock_base_channel)
+    wrapper1.assert_called_once_with(mock_chan_1)
+
+
+def test_apply_channel_wrappers_interspersed():
+    """Verify that a mixed sequence of gRPC interceptors and channel wrapper callables are applied."""
+
+    class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor):
+        def intercept_unary_unary(self, continuation, client_call_details, request):
+            return continuation(client_call_details, request)
+
+    interceptor = DummyUnaryInterceptor()
+    mock_base_channel = mock.Mock(name="base_channel")
+    mock_chan_after_wrapper = mock.Mock(name="chan_after_wrapper")
+    mock_chan_after_interceptor = mock.Mock(name="chan_after_interceptor")
+
+    wrapper = mock.Mock(return_value=mock_chan_after_wrapper)
+
+    with mock.patch(
+        "grpc.intercept_channel", return_value=mock_chan_after_interceptor
+    ) as mock_intercept:
+        result = grpc_helpers.apply_channel_wrappers(
+            mock_base_channel, [interceptor, wrapper]
+        )
+
+    assert result is mock_chan_after_interceptor
+    wrapper.assert_called_once_with(mock_base_channel)
+    mock_intercept.assert_called_once_with(mock_chan_after_wrapper, interceptor)
+
+
+def test_apply_channel_wrappers_invalid_type_raises():
+    """Verify that passing an invalid object that is neither an interceptor nor callable raises TypeError."""
+    mock_base_channel = mock.Mock(name="base_channel")
+    with pytest.raises(TypeError, match="Expected ChannelWrapper"):
+        grpc_helpers.apply_channel_wrappers(mock_base_channel, [12345])

From 7aeb5a2a43237ca93073e0facf4552adb01e39c1 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Mon, 31 Aug 2026 08:39:10 -0400
Subject: [PATCH 08/10] style(api-core): format imports with isort via ruff

---
 packages/google-api-core/google/api_core/grpc_helpers.py   | 1 +
 packages/google-api-core/tests/unit/test_client_options.py | 1 +
 packages/google-api-core/tests/unit/test_grpc_helpers.py   | 3 ++-
 packages/google-api-core/tests/unit/test_observability.py  | 1 +
 packages/google-api-core/tests/unit/test_path_template.py  | 1 +
 5 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 5f9baf4cf12f..7e7777766f7b 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -34,6 +34,7 @@
 import google.auth.transport.requests
 import google.protobuf
 import grpc
+
 from google.api_core import exceptions, general_helpers
 
 # The list of gRPC Callable interfaces that return iterators.
diff --git a/packages/google-api-core/tests/unit/test_client_options.py b/packages/google-api-core/tests/unit/test_client_options.py
index c15e83174ed4..49fb35e8ba2f 100644
--- a/packages/google-api-core/tests/unit/test_client_options.py
+++ b/packages/google-api-core/tests/unit/test_client_options.py
@@ -15,6 +15,7 @@
 from re import match
 
 import pytest
+
 from google.api_core import client_options
 
 from ..helpers import warn_deprecated_credentials_file
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 8d91e63f15f7..4367552651d7 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -24,9 +24,10 @@
     pytest.skip("No GRPC", allow_module_level=True)
 
 import google.auth.credentials
-from google.api_core import exceptions, grpc_helpers
 from google.longrunning import operations_pb2
 
+from google.api_core import exceptions, grpc_helpers
+
 
 def test__patch_callable_name():
     callable = mock.Mock(spec=["__class__"])
diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py
index d39edb806040..f0ebe0afc14d 100644
--- a/packages/google-api-core/tests/unit/test_observability.py
+++ b/packages/google-api-core/tests/unit/test_observability.py
@@ -16,6 +16,7 @@
 from unittest import mock
 
 import pytest
+
 from google.api_core import _observability
 from google.api_core._feature_gating_helpers import FeatureGatingError
 from google.api_core.client_options import ClientOptions
diff --git a/packages/google-api-core/tests/unit/test_path_template.py b/packages/google-api-core/tests/unit/test_path_template.py
index fb67973549f7..f053fc952176 100644
--- a/packages/google-api-core/tests/unit/test_path_template.py
+++ b/packages/google-api-core/tests/unit/test_path_template.py
@@ -18,6 +18,7 @@
 
 import pytest
 from google.api import auth_pb2
+
 from google.api_core import path_template
 
 

From 9149df8dda89fcd6c63088100b8bbc7b6dd6c9cb Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Mon, 31 Aug 2026 11:57:28 -0400
Subject: [PATCH 09/10] refactor(api-core): use PEP 604 union syntax in
 grpc_helpers.py and clarify reverse execution

---
 .../google/api_core/grpc_helpers.py            | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 7e7777766f7b..447d521bfd8f 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -24,7 +24,6 @@
     Optional,
     Sequence,
     TypeVar,
-    Union,
     get_args,
 )
 
@@ -44,12 +43,12 @@
 P = TypeVar("P")
 
 # Type alias representing any client-side gRPC interceptor
-ClientInterceptor = Union[
-    grpc.UnaryUnaryClientInterceptor,
-    grpc.UnaryStreamClientInterceptor,
-    grpc.StreamUnaryClientInterceptor,
-    grpc.StreamStreamClientInterceptor,
-]
+ClientInterceptor = (
+    grpc.UnaryUnaryClientInterceptor
+    | grpc.UnaryStreamClientInterceptor
+    | grpc.StreamUnaryClientInterceptor
+    | grpc.StreamStreamClientInterceptor
+)
 
 # Runtime tuple of gRPC client interceptor base classes for isinstance checks
 _CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
@@ -58,7 +57,7 @@
 ChannelWrapperCallable = Callable[[grpc.Channel], grpc.Channel]
 
 # Generic type alias representing any channel wrapper (interceptor or callable)
-ChannelWrapper = Union[ClientInterceptor, ChannelWrapperCallable]
+ChannelWrapper = ClientInterceptor | ChannelWrapperCallable
 
 
 def _patch_callable_name(callable_):
@@ -447,7 +446,7 @@ def _modify_target_for_direct_path(target: str) -> str:
 
 def apply_channel_wrappers(
     channel: grpc.Channel,
-    wrappers: Optional[Sequence[ChannelWrapper]] = None,
+    wrappers: Sequence[ChannelWrapper] | None = None,
 ) -> grpc.Channel:
     """Applies channel wrappers (client interceptors or channel-wrapping callables) to a gRPC channel.
 
@@ -472,6 +471,7 @@ def apply_channel_wrappers(
         return channel
 
     modified_channel = channel
+    # Reverse the inputs to align with the behavior of grpc.create_channel(*interceptors)
     for wrapper in reversed(list(wrappers)):
         if isinstance(wrapper, _CLIENT_INTERCEPTOR_CLASSES):
             modified_channel = grpc.intercept_channel(modified_channel, wrapper)

From f89cfbd7eda0b77fd838d0ad53e53b65f33e54e9 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Mon, 31 Aug 2026 12:05:11 -0400
Subject: [PATCH 10/10] fix(api-core): add TypeAlias and typing.cast to satisfy
 mypy

---
 .../google-api-core/google/api_core/grpc_helpers.py   | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 447d521bfd8f..13f89c64a13f 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -23,7 +23,9 @@
     Iterator,
     Optional,
     Sequence,
+    TypeAlias,
     TypeVar,
+    cast,
     get_args,
 )
 
@@ -43,7 +45,7 @@
 P = TypeVar("P")
 
 # Type alias representing any client-side gRPC interceptor
-ClientInterceptor = (
+ClientInterceptor: TypeAlias = (
     grpc.UnaryUnaryClientInterceptor
     | grpc.UnaryStreamClientInterceptor
     | grpc.StreamUnaryClientInterceptor
@@ -54,10 +56,10 @@
 _CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
 
 # Type alias representing a channel-wrapping callable
-ChannelWrapperCallable = Callable[[grpc.Channel], grpc.Channel]
+ChannelWrapperCallable: TypeAlias = Callable[[grpc.Channel], grpc.Channel]
 
 # Generic type alias representing any channel wrapper (interceptor or callable)
-ChannelWrapper = ClientInterceptor | ChannelWrapperCallable
+ChannelWrapper: TypeAlias = ClientInterceptor | ChannelWrapperCallable
 
 
 def _patch_callable_name(callable_):
@@ -476,7 +478,8 @@ def apply_channel_wrappers(
         if isinstance(wrapper, _CLIENT_INTERCEPTOR_CLASSES):
             modified_channel = grpc.intercept_channel(modified_channel, wrapper)
         elif callable(wrapper):
-            modified_channel = wrapper(modified_channel)
+            wrapper_callable = cast(ChannelWrapperCallable, wrapper)
+            modified_channel = wrapper_callable(modified_channel)
         else:
             raise TypeError(
                 f"Expected ChannelWrapper (ClientInterceptor or Callable[[Channel], Channel]), got {type(wrapper).__name__}"