diff --git a/agentplatform/agent_engines/templates/adk.py b/agentplatform/agent_engines/templates/adk.py index 9eaa49ce46..9be3e35b89 100644 --- a/agentplatform/agent_engines/templates/adk.py +++ b/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. diff --git a/tests/unit/agentplatform/frameworks/test_frameworks_adk.py b/tests/unit/agentplatform/frameworks/test_frameworks_adk.py index a3725842d9..38b38fa12a 100644 --- a/tests/unit/agentplatform/frameworks/test_frameworks_adk.py +++ b/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(): diff --git a/tests/unit/vertex_adk/test_agent_engine_templates_adk.py b/tests/unit/vertex_adk/test_agent_engine_templates_adk.py index 98078145dc..df8ea00596 100644 --- a/tests/unit/vertex_adk/test_agent_engine_templates_adk.py +++ b/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(): diff --git a/tests/unit/vertex_adk/test_reasoning_engine_templates_adk.py b/tests/unit/vertex_adk/test_reasoning_engine_templates_adk.py index 391dc7298a..c9daf249a2 100644 --- a/tests/unit/vertex_adk/test_reasoning_engine_templates_adk.py +++ b/tests/unit/vertex_adk/test_reasoning_engine_templates_adk.py @@ -18,6 +18,7 @@ import json import os import re +import threading from unittest import mock from typing import Optional @@ -27,6 +28,7 @@ from vertexai.agent_engines import _utils from vertexai.preview import reasoning_engines from vertexai.preview.reasoning_engines.templates import adk as adk_template +from google.genai import errors as genai_errors from google.genai import types import pytest import uuid @@ -659,6 +661,90 @@ def test_streaming_agent_run_with_events(self): events = list(app.streaming_agent_run_with_events(request_json=request_json)) assert len(events) == 1 + def test_streaming_agent_run_with_events_reraises_api_error(self): + """A mid-stream backend failure reaches the caller. + + `google.genai.errors.APIError` is not a `RuntimeError`, so the stream + used to just stop, which is indistinguishable from a short but + successful response (b/550103401). + """ + error = genai_errors.ServerError( + 500, {"error": {"message": "Internal error.", "status": "INTERNAL"}} + ) + + class _FailingRunner(_MockRunner): + def run(self, *args, **kwargs): + yield from super().run(*args, **kwargs) + raise error + + app = reasoning_engines.AdkApp( + agent=Agent(name=_TEST_AGENT_NAME, model=_TEST_MODEL) + ) + app.set_up() + app._tmpl_attrs[ # pylint: disable=protected-access + "in_memory_runner" + ] = _FailingRunner() + request_json = json.dumps( + { + "user_id": _TEST_USER_ID, + "message": { + "parts": [{"text": "What is the exchange rate from USD to SEK?"}], + "role": "user", + }, + } + ) + + with pytest.raises(genai_errors.ServerError) as exc_info: + list(app.streaming_agent_run_with_events(request_json=request_json)) + assert exc_info.value is error + + def test_streaming_agent_run_with_events_closed_early_after_error(self): + """Abandoning a failing stream does not deadlock. + + The worker thread queues exactly one terminal item. Queueing both the + exception and a `None` sentinel would park it forever on the + `maxsize=1` queue once the consumer stopped reading, taking the + generator's `thread.join()` down with it (b/550103401). + """ + error = genai_errors.ServerError( + 500, {"error": {"message": "Internal error.", "status": "INTERNAL"}} + ) + + class _FailingRunner(_MockRunner): + def run(self, *args, **kwargs): + yield from super().run(*args, **kwargs) + raise error + + app = reasoning_engines.AdkApp( + agent=Agent(name=_TEST_AGENT_NAME, model=_TEST_MODEL) + ) + app.set_up() + app._tmpl_attrs[ # pylint: disable=protected-access + "in_memory_runner" + ] = _FailingRunner() + request_json = json.dumps( + { + "user_id": _TEST_USER_ID, + "message": { + "parts": [{"text": "What is the exchange rate from USD to SEK?"}], + "role": "user", + }, + } + ) + + def _take_one_event_then_close(): + stream = app.streaming_agent_run_with_events(request_json=request_json) + for _ in stream: + break + stream.close() + + # Consume on a worker so a regression fails this test rather than + # hanging the whole suite. + consumer = threading.Thread(target=_take_one_event_then_close, daemon=True) + consumer.start() + consumer.join(timeout=60) + assert not consumer.is_alive(), "streaming generator deadlocked on close" + def test_streaming_agent_run_with_events_propagates_labels(self): from google.adk.agents.run_config import RunConfig @@ -1204,6 +1290,45 @@ 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 = reasoning_engines.AdkApp( + agent=Agent(name=_TEST_AGENT_NAME, model=_TEST_MODEL) + ) + 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.mark.asyncio async def test_bidi_stream_query_invalid_request_queue(self): app = reasoning_engines.AdkApp( diff --git a/vertexai/agent_engines/templates/adk.py b/vertexai/agent_engines/templates/adk.py index 5f814b5b12..ca4c1cc8b4 100644 --- a/vertexai/agent_engines/templates/adk.py +++ b/vertexai/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 ( @@ -577,6 +576,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 @@ -1494,34 +1527,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. @@ -1559,29 +1569,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, @@ -1645,35 +1635,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, @@ -1723,28 +1692,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. diff --git a/vertexai/preview/reasoning_engines/templates/adk.py b/vertexai/preview/reasoning_engines/templates/adk.py index 003812ab3a..df1a8bc9e9 100644 --- a/vertexai/preview/reasoning_engines/templates/adk.py +++ b/vertexai/preview/reasoning_engines/templates/adk.py @@ -638,6 +638,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 @@ -1246,11 +1280,13 @@ async def _invoke_agent_async(): def _asyncio_thread_main(): try: asyncio.run(_invoke_agent_async()) - except RuntimeError as e: - event_queue.put(e) - finally: # Use None as a sentinel to stop the main thread. event_queue.put(None) + except BaseException as e: # pylint: disable=broad-exception-caught + # Queue exactly one terminal item. The consumer stops on either + # of them, so a second put() on this maxsize=1 queue would block + # forever whenever the consumer abandoned the generator early. + event_queue.put(e) thread = threading.Thread(target=_asyncio_thread_main) thread.start() @@ -1260,7 +1296,7 @@ def _asyncio_thread_main(): event = event_queue.get() if event is None: break - if isinstance(event, RuntimeError): + if isinstance(event, BaseException): raise event yield event finally: @@ -1399,34 +1435,11 @@ def get_session( **kwargs, ): """Get a session for the given user.""" - 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 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( - "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. @@ -1451,29 +1464,9 @@ async def async_list_sessions(self, *, user_id: str, **kwargs): def list_sessions(self, *, user_id: str, **kwargs): """List sessions for the given user.""" - 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, @@ -1520,35 +1513,14 @@ def create_session( **kwargs, ): """Creates a new session.""" - 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, @@ -1585,28 +1557,11 @@ def delete_session( **kwargs, ): """Deletes a session for the given user.""" - 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.