fix: propagate the underlying API error from AdkApp's synchronous session methods and streaming agent runs

agentplatform/agent_engines/templates/adk.py

@@ -17,7 +17,6 @@
 from collections.abc import Awaitable
 import enum
 import os
-import queue
 import sys
 import threading
 from typing import (
@@ -593,6 +592,40 @@ def _override_active_span_processor(
     tracer_provider._active_span_processor = active_span_processor
 
 
+def _run_coroutine_on_thread(coroutine_fn: Callable[[], Awaitable[Any]]) -> Any:
+    """Runs a coroutine to completion on a dedicated worker thread.
+
+    The deprecated synchronous session methods cannot call `asyncio.run`
+    directly, because they may be invoked from a thread that already owns a
+    running event loop. Any exception raised by the coroutine is re-raised
+    here with its original traceback, so the caller sees the underlying
+    failure (e.g. a `google.genai.errors.APIError`) instead of a generic
+    error.
+
+    Args:
+        coroutine_fn (Callable[[], Awaitable[Any]]):
+            Required. A zero-argument callable returning the awaitable to run.
+            It is called on the worker thread.
+
+    Returns:
+        Any: The value returned by the awaitable.
+    """
+    outcome = {}
+
+    def _asyncio_thread_main():
+        try:
+            outcome["result"] = asyncio.run(coroutine_fn())
+        except BaseException as e:  # pylint: disable=broad-exception-caught
+            outcome["error"] = e
+
+    thread = threading.Thread(target=_asyncio_thread_main)
+    thread.start()
+    thread.join()
+    if "error" in outcome:
+        raise outcome["error"]
+    return outcome.get("result")
+
+
 def _validate_run_config(run_config: Optional[Dict[str, Any]]):
     """Validates the run config."""
     from google.adk.agents.run_config import RunConfig
@@ -1534,34 +1567,11 @@ def get_session(
             DeprecationWarning,
             stacklevel=2,
         )
-        event_queue = queue.Queue(maxsize=1)
-
-        async def _invoke_async_get_session():
-            return await self.async_get_session(
+        return _run_coroutine_on_thread(
+            lambda: self.async_get_session(
                 user_id=user_id, session_id=session_id, **kwargs
             )
-
-        def _asyncio_thread_main():
-            try:
-                result = asyncio.run(_invoke_async_get_session())
-                event_queue.put(result)
-            except Exception as e:
-                event_queue.put(e)
-
-        thread = threading.Thread(target=_asyncio_thread_main)
-        thread.start()
-
-        # Wait for the thread to finish
-        thread.join()
-        try:
-            outcome = event_queue.get(timeout=10)
-        except queue.Empty:
-            raise RuntimeError(
-                "Session not found. Please create it using .create_session()"
-            ) from None
-        if isinstance(outcome, RuntimeError):
-            raise outcome from None
-        return outcome
+        )
 
     async def async_list_sessions(self, *, user_id: str, **kwargs):
         """List sessions for the given user.
@@ -1599,29 +1609,9 @@ def list_sessions(self, *, user_id: str, **kwargs):
             DeprecationWarning,
             stacklevel=2,
         )
-        event_queue = queue.Queue()
-
-        async def _invoke_async_list_sessions():
-            try:
-                response = await self.async_list_sessions(user_id=user_id, **kwargs)
-                event_queue.put(response)
-            except RuntimeError as e:
-                event_queue.put(e)
-
-        def _asyncio_thread_main():
-            try:
-                asyncio.run(_invoke_async_list_sessions())
-            finally:
-                event_queue.put(None)
-
-        thread = threading.Thread(target=_asyncio_thread_main)
-        thread.start()
-        # Wait for the thread to finish
-        thread.join()
-        try:
-            return event_queue.get(timeout=10)
-        except queue.Empty:
-            raise RuntimeError("Failed to list sessions.") from None
+        return _run_coroutine_on_thread(
+            lambda: self.async_list_sessions(user_id=user_id, **kwargs)
+        )
 
     async def async_create_session(
         self,
@@ -1685,35 +1675,14 @@ def create_session(
             DeprecationWarning,
             stacklevel=2,
         )
-        event_queue = queue.Queue(maxsize=1)
-
-        async def _invoke_async_create_session():
-            return await self.async_create_session(
+        return _run_coroutine_on_thread(
+            lambda: self.async_create_session(
                 user_id=user_id,
                 session_id=session_id,
                 state=state,
                 **kwargs,
             )
-
-        def _asyncio_thread_main():
-            try:
-                result = asyncio.run(_invoke_async_create_session())
-                event_queue.put(result)
-            except RuntimeError as e:
-                event_queue.put(e)
-
-        thread = threading.Thread(target=_asyncio_thread_main)
-        thread.start()
-        # Wait for the thread to finish
-        thread.join()
-
-        try:
-            outcome = event_queue.get(timeout=10)
-        except queue.Empty:
-            raise RuntimeError("Failed to create session.") from None
-        if isinstance(outcome, RuntimeError):
-            raise outcome from None
-        return outcome
+        )
 
     async def async_delete_session(
         self,
@@ -1763,28 +1732,11 @@ def delete_session(
             DeprecationWarning,
             stacklevel=2,
         )
-        event_queue = queue.Queue(maxsize=1)
-
-        async def _invoke_async_delete_session():
-            await self.async_delete_session(
+        _run_coroutine_on_thread(
+            lambda: self.async_delete_session(
                 user_id=user_id, session_id=session_id, **kwargs
             )
-
-        def _asyncio_thread_main():
-            try:
-                asyncio.run(_invoke_async_delete_session())
-                event_queue.put(None)
-            except RuntimeError as e:
-                event_queue.put(e)
-
-        thread = threading.Thread(target=_asyncio_thread_main)
-        thread.start()
-        # Wait for the thread to finish
-        thread.join()
-
-        outcome = event_queue.get(timeout=10)
-        if isinstance(outcome, RuntimeError):
-            raise outcome from None
+        )
 
     async def async_add_session_to_memory(self, *, session: Dict[str, Any]):
         """Generates memories.

tests/unit/agentplatform/frameworks/test_frameworks_adk.py

@@ -39,6 +39,7 @@
 from agentplatform.agent_engines.templates import (
     adk as adk_template,
 )
+from google.genai import errors as genai_errors
 from google.genai import types
 import pytest
 import requests
@@ -1551,6 +1552,43 @@ async def test_async_stream_query_invalid_message_type(self):
             async for _ in app.async_stream_query(user_id=_TEST_USER_ID, message=123):
                 pass
 
+    @pytest.mark.parametrize(
+        "method_name,method_kwargs",
+        [
+            ("create_session", {"user_id": _TEST_USER_ID}),
+            (
+                "get_session",
+                {"user_id": _TEST_USER_ID, "session_id": "test_session_id"},
+            ),
+            ("list_sessions", {"user_id": _TEST_USER_ID}),
+            (
+                "delete_session",
+                {"user_id": _TEST_USER_ID, "session_id": "test_session_id"},
+            ),
+        ],
+    )
+    def test_sync_session_method_reraises_api_error(self, method_name, method_kwargs):
+        """A backend error reaches the caller instead of being swallowed.
+
+        `google.genai.errors.APIError` is not a `RuntimeError`, so it used to
+        escape the worker thread and leave the caller with a generic error, a
+        `None`, or the exception object itself (b/550103401).
+        """
+        app = adk_template.AdkApp(agent=_TEST_AGENT)
+        error = genai_errors.ServerError(
+            500, {"error": {"message": "Internal error.", "status": "INTERNAL"}}
+        )
+
+        async def _raise_server_error(*args, **kwargs):
+            raise error
+
+        with mock.patch.object(
+            adk_template.AdkApp, f"async_{method_name}", _raise_server_error
+        ):
+            with pytest.raises(genai_errors.ServerError) as exc_info:
+                getattr(app, method_name)(**method_kwargs)
+        assert exc_info.value is error
+
 
 @pytest.fixture(scope="module")
 def create_agent_engine_mock():

tests/unit/vertex_adk/test_agent_engine_templates_adk.py

@@ -38,6 +38,7 @@
 from vertexai.agent_engines import _agent_engines
 from vertexai.agent_engines import _utils
 from vertexai.agent_engines.templates import adk as adk_template
+from google.genai import errors as genai_errors
 from google.genai import types
 import pytest
 import requests
@@ -1266,6 +1267,43 @@ async def test_async_stream_query_invalid_message_type(self):
             async for _ in app.async_stream_query(user_id=_TEST_USER_ID, message=123):
                 pass
 
+    @pytest.mark.parametrize(
+        "method_name,method_kwargs",
+        [
+            ("create_session", {"user_id": _TEST_USER_ID}),
+            (
+                "get_session",
+                {"user_id": _TEST_USER_ID, "session_id": "test_session_id"},
+            ),
+            ("list_sessions", {"user_id": _TEST_USER_ID}),
+            (
+                "delete_session",
+                {"user_id": _TEST_USER_ID, "session_id": "test_session_id"},
+            ),
+        ],
+    )
+    def test_sync_session_method_reraises_api_error(self, method_name, method_kwargs):
+        """A backend error reaches the caller instead of being swallowed.
+
+        `google.genai.errors.APIError` is not a `RuntimeError`, so it used to
+        escape the worker thread and leave the caller with a generic error, a
+        `None`, or the exception object itself (b/550103401).
+        """
+        app = agent_engines.AdkApp(agent=_TEST_AGENT)
+        error = genai_errors.ServerError(
+            500, {"error": {"message": "Internal error.", "status": "INTERNAL"}}
+        )
+
+        async def _raise_server_error(*args, **kwargs):
+            raise error
+
+        with mock.patch.object(
+            adk_template.AdkApp, f"async_{method_name}", _raise_server_error
+        ):
+            with pytest.raises(genai_errors.ServerError) as exc_info:
+                getattr(app, method_name)(**method_kwargs)
+        assert exc_info.value is error
+
 
 @pytest.fixture(scope="module")
 def create_agent_engine_mock():