Fix gRPC client interceptor breaking bidirectional streaming (#1180)

Route bidi (stream-stream) RPCs through `_intercept` instead of the
generator-based `_intercept_server_stream`. The generator wrapper strips
the grpc.Call/grpc.Future interface, causing downstream code (e.g.
google.api_core.bidi.BidiRpc) to crash with:

    AttributeError: 'generator' object has no attribute 'add_done_callback'

The fix adds `and not client_info.is_client_stream` to the condition in
`intercept_stream()` so only unary-stream RPCs use the generator path.

Includes a regression test verifying the bidi stream response preserves
the grpc.Call interface (add_done_callback, cancel, is_active).

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: juandelacruz-calvo

CHANGELOG.md

@@ -59,6 +59,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4139](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4139))
 
 ### Fixed
+- `opentelemetry-instrumentation-grpc`: Fix bidirectional streaming RPCs raising `AttributeError: 'generator' object has no attribute 'add_done_callback'` by routing bidi streams through `_intercept` instead of the generator-based `_intercept_server_stream`
+  ([#1180](https://github.com/open-telemetry/opentelemetry-python-contrib/issues/1180))
 - `opentelemetry-instrumentation-mysql`: Refactor MySQL integration test mocks to use concrete DBAPI connection attributes, reducing noisy attribute type warnings.
   ([#4116](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4116))
 - `opentelemetry-instrumentation-cassandra`: Use `_instruments_any` instead of `_instruments` for driver dependencies so that having either `cassandra-driver` or `scylla-driver` installed is sufficient

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_client.py

@@ -228,7 +228,7 @@ def intercept_stream(
         if self._filter is not None and not self._filter(client_info):
             return invoker(request_or_iterator, metadata)
 
-        if client_info.is_server_stream:
+        if client_info.is_server_stream and not client_info.is_client_stream:
             return self._intercept_server_stream(
                 request_or_iterator, metadata, client_info, invoker
             )

instrumentation/opentelemetry-instrumentation-grpc/tests/test_client_interceptor.py

@@ -221,6 +221,51 @@ def test_stream_stream(self):
             },
         )
 
+    def test_stream_stream_preserves_call_interface(self):
+        """Regression test for issue #1180.
+
+        Bidirectional streaming RPCs must return an object that implements
+        grpc.Call (add_done_callback, cancel, is_active, etc.) rather than
+        a bare generator.  Before the fix, bidi streams were routed through
+        the generator-based _intercept_server_stream, which stripped the
+        grpc.Call interface and caused downstream code to crash with:
+            AttributeError: 'generator' object has no attribute 'add_done_callback'
+        """
+
+        def request_messages():
+            for _ in range(5):
+                yield Request(client_id=1, request_data="data")
+
+        response_iterator = self._stub.BidirectionalStreamingMethod(
+            request_messages(), metadata=(("key", "value"),)
+        )
+
+        for attr in ("add_done_callback", "cancel", "is_active"):
+            self.assertTrue(
+                hasattr(response_iterator, attr),
+                f"bidi stream response missing grpc.Call method '{attr}'",
+            )
+
+        list(response_iterator)
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        span = spans[0]
+
+        self.assertEqual(
+            span.name, "/GRPCTestServer/BidirectionalStreamingMethod"
+        )
+        self.assertIs(span.kind, trace.SpanKind.CLIENT)
+        self.assertSpanHasAttributes(
+            span,
+            {
+                RPC_METHOD: "BidirectionalStreamingMethod",
+                RPC_SERVICE: "GRPCTestServer",
+                RPC_SYSTEM: "grpc",
+                RPC_GRPC_STATUS_CODE: grpc.StatusCode.OK.value[0],
+            },
+        )
+
     def test_error_simple(self):
         with self.assertRaises(grpc.RpcError):
             simple_method(self._stub, error=True)