fix(backend/python): don't await sync servicer behaviors in AsyncModelIdentityInterceptor

The model-identity interceptor (added for #10952) is installed on every Python
backend's gRPC server. Its grpc.aio variant invokes the wrapped servicer
behavior itself and awaits the result unconditionally:

    result = await original(request, context)                 # LoadModel
    return await original_unary(request, context)             # guarded RPCs
    async for response in original_stream(request, context):  # streaming

But a backend's servicer methods may be plain sync functions. The transformers
backend, for one, defines `def LoadModel` and `def Embedding` (not `async def`).
grpc.aio's own dispatch adapts both shapes, but this interceptor calls the
behavior directly and bypasses that. For a sync method `original(...)` returns a
message object, not a coroutine, so the `await` raises:

    TypeError: object Result can't be used in 'await' expression

The model loads, then the LoadModel RPC dies on return; the guarded sync
Embedding fails the same way. It happens on every platform, not just one backend
build. CI never caught it because AsyncModelIdentityInterceptor had no
behavioral test -- only an "is it installed" assertion.

Fix: await only when the behavior actually returned an awaitable
(inspect.isawaitable), mirroring grpc.aio's own sync/async adaptation. The
streaming guard iterates a sync generator with `for` and an async one with
`async for`.

Adds async-path coverage to model_identity_test.py exercising both sync and
async LoadModel / guarded-unary / streaming behaviors. The sync cases fail on
the current code with the TypeError above and pass with this fix.

Signed-off-by: stefanwalcz <stefan.walcz@walcz.de>

Author: walcz-de

backend/python/common/model_identity.py

@@ -20,6 +20,7 @@
 the model itself.
 """
 
+import inspect
 import threading
 
 import grpc
@@ -200,7 +201,14 @@ async def intercept_service(self, continuation, handler_call_details):
             original = handler.unary_unary
 
             async def record(request, context):
-                result = await original(request, context)
+                # A backend's LoadModel may be a plain sync method (many define
+                # `def LoadModel`, not `async def`). grpc.aio's own dispatch
+                # adapts both, but this interceptor calls the behavior directly,
+                # so it must not await a non-awaitable return -- otherwise a sync
+                # backend fails with "object <T> can't be used in 'await'".
+                result = original(request, context)
+                if inspect.isawaitable(result):
+                    result = await result
                 if getattr(result, "success", True):
                     self.state.record(getattr(request, "Model", ""))
                 return result
@@ -214,8 +222,15 @@ async def guard_stream(request, context):
                 message = self.state.mismatch(getattr(request, "ModelIdentity", ""))
                 if message is not None:
                     await context.abort(grpc.StatusCode.NOT_FOUND, message)
-                async for response in original_stream(request, context):
-                    yield response
+                # A sync backend yields a plain generator, an async one an async
+                # generator; iterate whichever this is.
+                stream = original_stream(request, context)
+                if hasattr(stream, "__aiter__"):
+                    async for response in stream:
+                        yield response
+                else:
+                    for response in stream:
+                        yield response
 
             return _rebuild(handler, guard_stream)
 
@@ -225,6 +240,9 @@ async def guard(request, context):
             message = self.state.mismatch(getattr(request, "ModelIdentity", ""))
             if message is not None:
                 await context.abort(grpc.StatusCode.NOT_FOUND, message)
-            return await original_unary(request, context)
+            result = original_unary(request, context)
+            if inspect.isawaitable(result):
+                result = await result
+            return result
 
         return _rebuild(handler, guard)

backend/python/common/model_identity_test.py

@@ -9,6 +9,7 @@
 enforcement that rejects requests it should serve.
 """
 
+import asyncio
 import os
 import unittest
 
@@ -302,6 +303,109 @@ def test_every_modality_rpc_serves_without_an_identity(self):
     def test_codec_rpcs_stay_unguarded(self):
         for method in ("/backend.Backend/AudioEncode", "/backend.Backend/AudioDecode"):
             self.assertNotIn(method, model_identity._GUARDED_METHODS)
+class TestAsyncInterceptorBehavior(unittest.TestCase):
+    """The grpc.aio counterpart, which had no behavioral coverage.
+
+    AsyncModelIdentityInterceptor wraps a backend's own servicer behavior. That
+    behavior may be sync or async: several backends define `def LoadModel` and
+    `def Embedding` (not `async def`), and grpc.aio's dispatch adapts both. The
+    interceptor invokes the behavior itself, so if it awaits unconditionally it
+    breaks every sync method it guards with "object <T> can't be used in
+    'await'". These tests exercise both shapes; the sync ones are the regression.
+    """
+
+    def setUp(self):
+        self.interceptor = model_identity.AsyncModelIdentityInterceptor()
+
+    def _wrap(self, method, handler):
+        async def continuation(_):
+            return handler
+
+        return asyncio.run(
+            self.interceptor.intercept_service(continuation, _FakeCallDetails(method))
+        )
+
+    def _load(self, behavior, model):
+        wrapped = self._wrap("/backend.Backend/LoadModel", _handler(behavior))
+        return asyncio.run(wrapped.unary_unary(_Request(Model=model), _FakeContext()))
+
+    def _call_unary(self, behavior, identity):
+        wrapped = self._wrap("/backend.Backend/Predict", _handler(behavior))
+        return asyncio.run(
+            wrapped.unary_unary(_Request(ModelIdentity=identity), _FakeContext())
+        )
+
+    def _drain_stream(self, behavior, identity):
+        wrapped = self._wrap(
+            "/backend.Backend/PredictStream", _handler(behavior, response_streaming=True)
+        )
+
+        async def drain():
+            out = []
+            async for item in wrapped.unary_stream(
+                _Request(ModelIdentity=identity), _FakeContext()
+            ):
+                out.append(item)
+            return out
+
+        return asyncio.run(drain())
+
+    # --- LoadModel: sync behavior is the regression, async must still work ---
+
+    def test_load_records_with_sync_behavior(self):
+        self._load(lambda request, context: _Result(), "a.gguf")
+        self.assertEqual(self.interceptor.state.loaded, "a.gguf")
+
+    def test_load_records_with_async_behavior(self):
+        async def behavior(request, context):
+            return _Result()
+
+        self._load(behavior, "a.gguf")
+        self.assertEqual(self.interceptor.state.loaded, "a.gguf")
+
+    def test_failed_sync_load_records_nothing(self):
+        self._load(lambda request, context: _Result(success=False), "a.gguf")
+        self.assertEqual(self.interceptor.state.loaded, "")
+
+    # --- guarded unary: sync and async behaviors both served / rejected ---
+
+    def test_guard_serves_sync_behavior(self):
+        self._load(lambda request, context: _Result(), "a.gguf")
+        result = self._call_unary(lambda request, context: "served", "a.gguf")
+        self.assertEqual(result, "served")
+
+    def test_guard_serves_async_behavior(self):
+        self._load(lambda request, context: _Result(), "a.gguf")
+
+        async def behavior(request, context):
+            return "served"
+
+        self.assertEqual(self._call_unary(behavior, "a.gguf"), "served")
+
+    def test_guard_rejects_mismatch(self):
+        self._load(lambda request, context: _Result(), "a.gguf")
+        with self.assertRaises(_Aborted):
+            self._call_unary(lambda request, context: "served", "b.gguf")
+
+    # --- guarded stream: sync generator and async generator both work ---
+
+    def test_guard_stream_serves_sync_generator(self):
+        self._load(lambda request, context: _Result(), "a.gguf")
+
+        def behavior(request, context):
+            yield "a"
+            yield "b"
+
+        self.assertEqual(self._drain_stream(behavior, "a.gguf"), ["a", "b"])
+
+    def test_guard_stream_serves_async_generator(self):
+        self._load(lambda request, context: _Result(), "a.gguf")
+
+        async def behavior(request, context):
+            yield "a"
+            yield "b"
+
+        self.assertEqual(self._drain_stream(behavior, "a.gguf"), ["a", "b"])
 
 
 if __name__ == "__main__":