From 48087970cd39db43f8cb199268d86cd7dda67a3b Mon Sep 17 00:00:00 2001 From: Lam Nguyen Date: Wed, 8 Jul 2026 11:24:07 +0700 Subject: [PATCH 1/4] fix(timeline): make __repr__ recursion-safe to stop tracing cascade MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Timeline.__repr__ eagerly deep-formatted the last 10 entries, including metadata that can hold cyclic or non-serializable tool/resource objects. That sent repr() into infinite recursion whenever a tracer (LangSmith str() fallback on a non-serializable CompressedTimeline) or an error handler touched the timeline. The cascade exhausted the interpreter's recursion budget and crashed build_prompt — inspect.signature (recursive internally) died first as a victim, masking the real underlying error. Bounded __repr__ to entries count + token budget; CompressedTimeline inherits it. Detail remains available via get_timeline_summary()/to_dict(). Also harden Misc.parse_method_signature: wrap inspect.signature so a pathological __wrapped__ chain or low recursion budget falls back to follow_wrapped=False instead of failing build_prompt on one bad method. Adds two regression tests (cyclic entry metadata, large timeline). --- dana/common/utils/misc.py | 13 ++++++++- dana/core/timeline/timeline.py | 17 ++++++++---- tests/unit/test_timeline.py | 48 ++++++++++++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 6 deletions(-) diff --git a/dana/common/utils/misc.py b/dana/common/utils/misc.py index c85ffde..46952e3 100644 --- a/dana/common/utils/misc.py +++ b/dana/common/utils/misc.py @@ -245,7 +245,18 @@ def parse_method_signature(method: callable, object_id: str | None = None) -> Me class_name = parts[-2] # Get class name before method name # If class_name is still None at this point, it's a standalone function - sig = inspect.signature(method) + # Defensive signature introspection. inspect.signature is recursive + # internally and can blow up on pathological __wrapped__ chains, or + # when the interpreter's recursion budget is already exhausted (e.g. + # after a tracing cascade). build_prompt must not die on one bad + # resource method — fall back to no-follow, then re-raise the original. + try: + sig = inspect.signature(method) + except (RecursionError, ValueError, TypeError) as _sig_err: + try: + sig = inspect.signature(method, follow_wrapped=False) + except Exception: + raise _sig_err docstring = inspect.getdoc(method) or "" # Parse docstring sections diff --git a/dana/core/timeline/timeline.py b/dana/core/timeline/timeline.py index ef2246e..a2e8487 100644 --- a/dana/core/timeline/timeline.py +++ b/dana/core/timeline/timeline.py @@ -488,13 +488,20 @@ def __init__( self._repository = None def __repr__(self) -> str: - """ - Return a string representation of the timeline. + """Return a bounded, recursion-safe string representation. - Returns: - String representation of the timeline + Entries hold arbitrary objects in ``metadata`` and ``content`` + (tool/resource payloads) that can be cyclic or non-serializable. + Eagerly deep-formatting them sends ``repr()`` into infinite + recursion, which crashes tracers (LangSmith ``str()`` fallback) + and every error handler that logs the timeline. Keep this cheap + and safe; use ``get_timeline_summary()`` / ``to_dict()`` for detail. """ - return f"Timeline(max_context_tokens={self.max_context_tokens}, timeline={self.timeline[-10:]})" + try: + entry_count = len(self.timeline) + except Exception: + entry_count = -1 + return f"Timeline(entries={entry_count}, max_context_tokens={self.max_context_tokens})" def add_entry(self, entry: TimelineEntry) -> None: """ diff --git a/tests/unit/test_timeline.py b/tests/unit/test_timeline.py index b0783cb..64181c1 100644 --- a/tests/unit/test_timeline.py +++ b/tests/unit/test_timeline.py @@ -685,3 +685,51 @@ def test_timeline_save_and_read_full_tool_sequence(self): # Verify final response assert read_entries[3].entry_type == TimelineEntryType.AGENT_RESPONSE assert "72F" in read_entries[3].content + + +class TestTimelineReprSafety: + """Regression tests for recursion-safe __repr__. + + Entries hold arbitrary objects in metadata (tool/resource payloads). A + cyclic or non-serializable object there used to make repr() recurse + infinitely, which crashed LangSmith tracing (str() fallback) and every + error handler that logged the timeline. repr() must stay bounded. + """ + + def test_repr_with_cyclic_entry_metadata_does_not_recurse(self): + from dataclasses import dataclass + from typing import Any + + @dataclass + class _CyclicMeta: + name: str = "payload" + back: Any = None + + cyclic = _CyclicMeta() + cyclic.back = cyclic # self-reference → dataclass repr would recurse + + entry = TimelineEntry( + entry_type=TimelineEntryType.RESOURCE_RESULT, + content="ok", + metadata={"payload": cyclic}, + ) + timeline = Timeline(max_context_tokens=1000) + timeline.add_entry(entry) + + # Must not raise RecursionError; must be a bounded string. + rendered = repr(timeline) + assert isinstance(rendered, str) + assert len(rendered) < 500 + assert "Timeline" in rendered + assert "entries=1" in rendered + + def test_repr_is_stable_for_large_timeline(self): + timeline = Timeline(max_context_tokens=1000) + for i in range(500): + timeline.add_entry(TimelineEntry(entry_type=TimelineEntryType.USER_MESSAGE, content=f"msg {i}")) + + rendered = repr(timeline) + # Bounded regardless of entry count — no deep formatting. + assert isinstance(rendered, str) + assert len(rendered) < 500 + assert "entries=500" in rendered From 1b4923420891d5bebd7c6dd072fbb3f096906705 Mon Sep 17 00:00:00 2001 From: Lam Nguyen Date: Wed, 8 Jul 2026 12:04:27 +0700 Subject: [PATCH 2/4] fix(llm): bridge chat_response_sync to a background loop under async MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chat_response_sync ran the async chat_response via loop.run_until_complete, which calls _check_running() and raises "Cannot run the event loop while another loop is running" whenever the caller already has a running loop on its thread — i.e. any sync agent API call (query/compression/reactive_compact) made from inside async code. The pre-existing async-context guard was dead: its own `raise RuntimeError` was caught by the surrounding `except RuntimeError: pass`, so the cryptic asyncio error surfaced instead. Fix: detect a running loop; when one is present, ship the coroutine to a dedicated background event loop on a daemon thread via run_coroutine_threadsafe and block on the result. The background loop spins independently of the caller's thread, so the coroutine progresses while the caller blocks — no deadlock. When no loop is running, the existing run_until_complete path is unchanged. Trade-off: under a running loop the caller's loop is blocked for the call duration. Acceptable for interactive/single-tenant use (the reported case); concurrent servers should use the async chat_response()/aquery() path. Adds three regression tests (no-loop, inside-running-loop, exception propagation through the bridge). --- dana/common/llm/llm.py | 57 ++++++++++++--- .../test_llm_chat_response_sync_bridge.py | 69 +++++++++++++++++++ 2 files changed, 116 insertions(+), 10 deletions(-) create mode 100644 tests/unit/test_llm_chat_response_sync_bridge.py diff --git a/dana/common/llm/llm.py b/dana/common/llm/llm.py index 56ed792..5cb79b2 100644 --- a/dana/common/llm/llm.py +++ b/dana/common/llm/llm.py @@ -29,6 +29,7 @@ def get_logger() -> logging.Logger: import asyncio import atexit import os +import threading import time from dana.common.observable import observable @@ -58,6 +59,35 @@ def _get_or_create_event_loop() -> asyncio.AbstractEventLoop: return _sync_event_loop +# Dedicated background event loop running in a daemon thread. Used to bridge +# sync→async calls when the caller already has a running loop on its own +# thread: ``run_until_complete`` refuses to nest inside a running loop, so we +# ship the coroutine here via ``run_coroutine_threadsafe`` and block on it. +_background_loop: asyncio.AbstractEventLoop | None = None +_background_loop_lock = threading.Lock() + + +def _get_or_create_background_loop() -> asyncio.AbstractEventLoop: + """Return a singleton event loop running in a daemon thread. + + The background loop spins independently of the caller's thread, so a + coroutine scheduled on it makes progress even while the caller blocks on + ``future.result()`` — no deadlock. + """ + global _background_loop + with _background_loop_lock: + if _background_loop is None or _background_loop.is_closed(): + loop = asyncio.new_event_loop() + thread = threading.Thread( + target=loop.run_forever, + daemon=True, + name="dana-sync-async-bridge", + ) + thread.start() + _background_loop = loop + return _background_loop + + def _cleanup_event_loop(): """Clean up the persistent event loop on exit.""" global _sync_event_loop @@ -295,20 +325,27 @@ def chat_response_sync(self, messages: list[LLMMessage], **kwargs) -> LLMRespons ValueError: If messages list is empty ProviderError: If the provider operation fails """ - # Check if we're already in an async context + # Bridge sync→async. Two calling contexts: + # 1) No loop running on this thread → drive the coroutine on the + # persistent sync loop directly. + # 2) A loop is already running on this thread (caller invoked the sync + # API from inside async code, e.g. via a framework that mixes + # sync/async). run_until_complete cannot nest inside a running loop, + # so ship the coroutine to a dedicated background loop on a daemon + # thread and block on the result. The background loop runs in its own + # thread, so the coroutine progresses while this thread blocks — no + # deadlock. Caller's loop is blocked for the call's duration; callers + # that need concurrency should use the async chat_response() instead. + coro = self.chat_response(messages, **kwargs) try: asyncio.get_running_loop() - # We're in an async context, but this is a sync method - # This should not normally happen, but if it does, raise an error - raise RuntimeError("chat_response_sync() cannot be called from within an async context. Use chat_response() instead.") except RuntimeError: - # No running loop, which is expected for sync method - pass + loop = _get_or_create_event_loop() + return loop.run_until_complete(coro) - # Reuse a persistent event loop to avoid "Event loop is closed" errors - # when async HTTP clients (httpx) try to close connections after asyncio.run() closes the loop - loop = _get_or_create_event_loop() - return loop.run_until_complete(self.chat_response(messages, **kwargs)) + bg_loop = _get_or_create_background_loop() + future = asyncio.run_coroutine_threadsafe(coro, bg_loop) + return future.result() async def ask(self, question: str, system_prompt: str | None = None, **kwargs) -> str: """ diff --git a/tests/unit/test_llm_chat_response_sync_bridge.py b/tests/unit/test_llm_chat_response_sync_bridge.py new file mode 100644 index 0000000..813b29c --- /dev/null +++ b/tests/unit/test_llm_chat_response_sync_bridge.py @@ -0,0 +1,69 @@ +"""Regression tests for chat_response_sync sync→async bridging. + +When a caller invokes the sync LLM API from inside a running asyncio loop +(e.g. a framework that mixes sync/async), ``run_until_complete`` cannot +nest inside that loop. ``chat_response_sync`` must bridge via a dedicated +background daemon-thread loop instead of raising +"Cannot run the event loop while another loop is running". +""" + +import asyncio + +import pytest + +from dana.common.llm.llm import LLM +from dana.common.llm.types import LLMMessage, LLMResponse + + +def _llm_returning(response: LLMResponse, *, raise_exc: Exception | None = None) -> LLM: + """Build an LLM whose async ``chat_response`` returns ``response`` (or raises).""" + llm = LLM() + + async def fake_chat_response(messages, **kwargs): + if raise_exc is not None: + raise raise_exc + return response + + llm.chat_response = fake_chat_response # type: ignore[assignment,method-assign] + return llm + + +def _user_msg() -> list[LLMMessage]: + return [LLMMessage(role="user", content="hi")] + + +def test_sync_works_without_running_loop(): + """Baseline: plain sync call uses the run_until_complete path.""" + expected = LLMResponse(content="ok", model="test-model") + llm = _llm_returning(expected) + + result = llm.chat_response_sync(_user_msg()) + assert result is expected + + +def test_sync_works_inside_running_loop(): + """Regression: sync API called from within a running loop must not crash. + + Previously raised: + RuntimeError: Cannot run the event loop while another loop is running + """ + expected = LLMResponse(content="ok", model="test-model") + llm = _llm_returning(expected) + + async def caller(): + # We are now inside a running loop on this thread. + return llm.chat_response_sync(_user_msg()) + + result = asyncio.run(caller()) + assert result is expected + + +def test_sync_propagates_exception_inside_running_loop(): + """Errors raised by the coroutine surface to the sync caller via the bridge.""" + llm = _llm_returning(LLMResponse(content="x", model="m"), raise_exc=RuntimeError("boom")) + + async def caller(): + return llm.chat_response_sync(_user_msg()) + + with pytest.raises(RuntimeError, match="boom"): + asyncio.run(caller()) From 9de161c1ffdee0d1b4e1dc0e8fa3f93d41b356c7 Mon Sep 17 00:00:00 2001 From: Lam Nguyen Date: Wed, 8 Jul 2026 12:35:18 +0700 Subject: [PATCH 3/4] refactor(llm): replace chat_response_sync thread bridge with loud guard The async path (chat_response / agent.aquery / _think_async / call_llm_async) is complete and is the correct entry point from async code. Bridging inside chat_response_sync via a background daemon-thread loop was the wrong layer: it masked sync-from-async misuse and blocked the caller's loop, instead of routing callers onto the async interface. Revert the background-loop bridge. chat_response_sync now fails loud when called from within a running event loop, with guidance toward the async API. This also fixes the pre-existing dead guard (its `raise RuntimeError` was caught by the surrounding `except RuntimeError: pass`, so the cryptic "Cannot run the event loop while another loop is running" surfaced instead). Legit sync callers are unaffected: sync query() from a sync context and the reflection daemon thread both run with no event loop on their thread, so they take the run_until_complete path as before. Tests updated to assert the loud-guard behavior (raises with guidance) plus the no-loop baseline and exception propagation. --- dana/common/llm/llm.py | 67 ++++++------------- .../test_llm_chat_response_sync_bridge.py | 50 +++++++------- 2 files changed, 47 insertions(+), 70 deletions(-) diff --git a/dana/common/llm/llm.py b/dana/common/llm/llm.py index 5cb79b2..31d0a65 100644 --- a/dana/common/llm/llm.py +++ b/dana/common/llm/llm.py @@ -29,7 +29,6 @@ def get_logger() -> logging.Logger: import asyncio import atexit import os -import threading import time from dana.common.observable import observable @@ -59,35 +58,6 @@ def _get_or_create_event_loop() -> asyncio.AbstractEventLoop: return _sync_event_loop -# Dedicated background event loop running in a daemon thread. Used to bridge -# sync→async calls when the caller already has a running loop on its own -# thread: ``run_until_complete`` refuses to nest inside a running loop, so we -# ship the coroutine here via ``run_coroutine_threadsafe`` and block on it. -_background_loop: asyncio.AbstractEventLoop | None = None -_background_loop_lock = threading.Lock() - - -def _get_or_create_background_loop() -> asyncio.AbstractEventLoop: - """Return a singleton event loop running in a daemon thread. - - The background loop spins independently of the caller's thread, so a - coroutine scheduled on it makes progress even while the caller blocks on - ``future.result()`` — no deadlock. - """ - global _background_loop - with _background_loop_lock: - if _background_loop is None or _background_loop.is_closed(): - loop = asyncio.new_event_loop() - thread = threading.Thread( - target=loop.run_forever, - daemon=True, - name="dana-sync-async-bridge", - ) - thread.start() - _background_loop = loop - return _background_loop - - def _cleanup_event_loop(): """Clean up the persistent event loop on exit.""" global _sync_event_loop @@ -325,27 +295,30 @@ def chat_response_sync(self, messages: list[LLMMessage], **kwargs) -> LLMRespons ValueError: If messages list is empty ProviderError: If the provider operation fails """ - # Bridge sync→async. Two calling contexts: - # 1) No loop running on this thread → drive the coroutine on the - # persistent sync loop directly. - # 2) A loop is already running on this thread (caller invoked the sync - # API from inside async code, e.g. via a framework that mixes - # sync/async). run_until_complete cannot nest inside a running loop, - # so ship the coroutine to a dedicated background loop on a daemon - # thread and block on the result. The background loop runs in its own - # thread, so the coroutine progresses while this thread blocks — no - # deadlock. Caller's loop is blocked for the call's duration; callers - # that need concurrency should use the async chat_response() instead. - coro = self.chat_response(messages, **kwargs) + # Fail loud when called from within a running event loop. The sync API + # cannot nest inside a running loop (run_until_complete refuses), and + # silently bridging via a background thread would mask the misuse and + # block the caller's loop. The async path (chat_response / agent.aquery) + # is the correct entry point from async code. + # + # The previous guard here was dead: its explicit `raise RuntimeError` + # was caught by the surrounding `except RuntimeError: pass`, so callers + # got the cryptic "Cannot run the event loop while another loop is + # running" instead of guidance. try: asyncio.get_running_loop() except RuntimeError: - loop = _get_or_create_event_loop() - return loop.run_until_complete(coro) + pass # No running loop on this thread — sync path is valid here. + else: + raise RuntimeError( + "chat_response_sync() was called from within a running event loop. " + "Use 'await llm.chat_response()' (or 'await agent.aquery()') instead. " + "If you reached here from a sync REPL, ensure it routes through the " + "async converse/aquery path." + ) - bg_loop = _get_or_create_background_loop() - future = asyncio.run_coroutine_threadsafe(coro, bg_loop) - return future.result() + loop = _get_or_create_event_loop() + return loop.run_until_complete(self.chat_response(messages, **kwargs)) async def ask(self, question: str, system_prompt: str | None = None, **kwargs) -> str: """ diff --git a/tests/unit/test_llm_chat_response_sync_bridge.py b/tests/unit/test_llm_chat_response_sync_bridge.py index 813b29c..92bfe34 100644 --- a/tests/unit/test_llm_chat_response_sync_bridge.py +++ b/tests/unit/test_llm_chat_response_sync_bridge.py @@ -1,10 +1,16 @@ -"""Regression tests for chat_response_sync sync→async bridging. - -When a caller invokes the sync LLM API from inside a running asyncio loop -(e.g. a framework that mixes sync/async), ``run_until_complete`` cannot -nest inside that loop. ``chat_response_sync`` must bridge via a dedicated -background daemon-thread loop instead of raising -"Cannot run the event loop while another loop is running". +"""Regression tests for chat_response_sync loop handling. + +``chat_response_sync`` runs the async ``chat_response`` via +``loop.run_until_complete``, which cannot nest inside a running event loop. +Rather than silently bridging (which would mask misuse and block the caller's +loop), the sync API must FAIL LOUD with guidance toward the async path +(``await llm.chat_response()`` / ``await agent.aquery()``) when called from +within a running loop. + +The previous guard here was dead — its explicit ``raise RuntimeError`` was +caught by the surrounding ``except RuntimeError: pass``, so callers got the +cryptic "Cannot run the event loop while another loop is running" instead of +the actionable message. """ import asyncio @@ -33,7 +39,7 @@ def _user_msg() -> list[LLMMessage]: def test_sync_works_without_running_loop(): - """Baseline: plain sync call uses the run_until_complete path.""" + """Baseline: plain sync call (no running loop) uses run_until_complete.""" expected = LLMResponse(content="ok", model="test-model") llm = _llm_returning(expected) @@ -41,29 +47,27 @@ def test_sync_works_without_running_loop(): assert result is expected -def test_sync_works_inside_running_loop(): - """Regression: sync API called from within a running loop must not crash. +def test_sync_raises_loud_inside_running_loop(): + """Sync API called from within a running loop must raise a clear, actionable error. - Previously raised: - RuntimeError: Cannot run the event loop while another loop is running + Previously: the dead guard let it fall through to run_until_complete → + "Cannot run the event loop while another loop is running" (cryptic). """ - expected = LLMResponse(content="ok", model="test-model") - llm = _llm_returning(expected) + llm = _llm_returning(LLMResponse(content="ok", model="m")) async def caller(): - # We are now inside a running loop on this thread. return llm.chat_response_sync(_user_msg()) - result = asyncio.run(caller()) - assert result is expected + with pytest.raises(RuntimeError) as exc_info: + asyncio.run(caller()) + msg = str(exc_info.value) + assert "chat_response_sync" in msg + assert "aquery" in msg or "chat_response" in msg -def test_sync_propagates_exception_inside_running_loop(): - """Errors raised by the coroutine surface to the sync caller via the bridge.""" +def test_sync_propagates_exception_without_running_loop(): + """Coroutine errors surface to the sync caller via run_until_complete.""" llm = _llm_returning(LLMResponse(content="x", model="m"), raise_exc=RuntimeError("boom")) - async def caller(): - return llm.chat_response_sync(_user_msg()) - with pytest.raises(RuntimeError, match="boom"): - asyncio.run(caller()) + llm.chat_response_sync(_user_msg()) From bff7d501088a14e5786722f7dc6d7bfd0a255c3c Mon Sep 17 00:00:00 2001 From: Lam Nguyen Date: Wed, 8 Jul 2026 12:35:20 +0700 Subject: [PATCH 4/4] feat(communicator): route sync converse through the async STAR path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit converse() (the sync interactive REPL entry) now delegates to aconverse(), which awaits agent.aquery() — so REPL turns use the fully-async STAR path (aquery → _think_async → call_llm_async → chat_response) instead of the sync query()/chat_response_sync path that crashed under a running event loop. _run_coroutine_blocking() runs aconverse via asyncio.run when no loop is running, and on a short-lived worker thread when a loop is already running, so the sync entry point remains usable from any caller. The bridge lives at the REPL entry (one place), keeping everything below it purely async. Adds test_communicator_converse_async.py covering the sync-wrapper-runs- async-path-from-running-loop regression. --- dana/core/agent/components/communicator.py | 206 +++--------------- .../core/test_communicator_converse_async.py | 42 ++++ 2 files changed, 69 insertions(+), 179 deletions(-) create mode 100644 tests/unit/core/test_communicator_converse_async.py diff --git a/dana/core/agent/components/communicator.py b/dana/core/agent/components/communicator.py index 71bda67..a1e8a41 100644 --- a/dana/core/agent/components/communicator.py +++ b/dana/core/agent/components/communicator.py @@ -8,6 +8,7 @@ import asyncio from collections.abc import Awaitable, Callable +import threading from typing import TYPE_CHECKING from uuid import uuid4 @@ -39,189 +40,14 @@ def converse(self, initial_message: str | None = None, session_id: str | None = """ Interactive conversation loop with a human user. + This sync entrypoint is a compatibility wrapper over ``aconverse`` so + REPL turns use the async STAR path (``aquery`` / async LLM calls). + Args: initial_message: Optional initial message to start the conversation session_id: Optional session identifier. If None, generates UUID. """ - # Generate session_id if not provided - if session_id is None: - session_id = str(uuid4()) - - agent_type = self._agent.agent_type - print(f"\n=== {agent_type.upper()} AGENT CONVERSATION ===") - print("Type '/quit', '/exit', or '/bye' to end the conversation") - print("Type '/help' for available commands") - print("=" * 50) - - # Track if we should use initial_message on first iteration - first_iteration = True - - while True: - try: - # Get user input (use initial_message on first iteration if provided) - if first_iteration and initial_message: - user_input = initial_message - print(f"\nYou: {user_input}") - first_iteration = False - else: - user_input = input("\nYou: ").strip() - # Save events if EventLog exists - if hasattr(self._agent, "_event_log") and self._agent._event_log is not None: - self._agent._event_log.save(session_id) - # Save timeline (agent, codec, storage_config already set in __init__) - if hasattr(self._agent, "_timeline") and self._agent._timeline is not None: - self._agent._timeline.save(session_id) - - # Check for exit commands - if user_input.lower() in ["/quit", "/exit", "/bye", "/q"]: - print("\nAgent: Goodbye! Thanks for the conversation.") - break - - # Check for help command - if user_input.lower() == "/help": - print("\n=== AVAILABLE COMMANDS ===") - print("• /quit, /exit, /bye, /q - End conversation") - print("• /help - Show this help") - print("• /timeline - Show conversation timeline") - print("• /state - Show agent state") - print("• /resources - List available resources") - print("• /workflows - List available workflows") - print("• /agents - List available agents") - print("• @agent_name/@agent_id message - Send direct message to specific agent") - print("• Any other text - Send message to agent") - continue - - # Check for special commands - if user_input.lower() == "/timeline": - print("\n=== CONVERSATION TIMELINE ===") - print(self._agent._state.get_timeline_summary()) - continue - - if user_input.lower() == "/state": - print("\n=== AGENT STATE ===") - state = self._agent._state.get_state() - for key, value in state.items(): - print(f"{key}: {value}") - continue - - if user_input.lower() == "/resources": - resources = self._agent.available_resources - print("\n=== AVAILABLE RESOURCES ===") - if resources: - for resource in resources: - print(f"• {resource.resource_type} (ID: {resource.resource_id})") - else: - print("No resources available") - continue - - if user_input.lower() == "/workflows": - workflows = self._agent.available_workflows - print("\n=== AVAILABLE WORKFLOWS ===") - if workflows: - for workflow in workflows: - print(f"• {workflow.workflow_type} (ID: {workflow.workflow_id})") - else: - print("No workflows available") - continue - - if user_input.lower() == "/agents": - agents = self._agent.available_agents - print("\n=== AVAILABLE AGENTS ===") - if agents: - for agent in agents: - print(f"• {agent.agent_type} (ID: {agent.object_id})") - else: - print("No other agents available") - continue - - # Check for direct agent messages (@agent_name message) - if user_input.startswith("@"): - # Parse @agent_name and message - parts = user_input[1:].split(" ", 1) - if len(parts) < 2: - print(f"\nInvalid format: {user_input}") - print("Use: @agent_name/@agent_id your message here") - continue - - target_agent_name = parts[0] - message = parts[1] - - # Find the target agent - target_agent = None - # Include current agent in the search list - all_agents = list(self._agent.available_agents) + [self._agent] - for agent in all_agents: - if agent.agent_type.lower() == target_agent_name.lower() or agent.object_id == target_agent_name: - target_agent = agent - break - - if target_agent is None: - print(f"\nAgent '{target_agent_name}' not found") - print("Type '/agents' to see available agents and their IDs") - continue - - # Send message to target agent - print(f"\nSending to {target_agent.agent_type}: ", end="", flush=True) - traces = target_agent.query(message=message, session_id=session_id) - response = traces.get("response", "No response generated") - print(response) - continue - - # Check for unrecognized commands (start with / but not recognized) - if user_input.startswith("/") and user_input.lower() not in [ - "/quit", - "/exit", - "/bye", - "/q", - "/help", - "/timeline", - "/state", - "/resources", - "/workflows", - "/agents", - ]: - print(f"\nCommand not supported: {user_input}") - print("Type '/help' for available commands") - continue - - # Skip empty input - if not user_input: - continue - - # Process the message through the agent - print("\nAgent: ", end="", flush=True) - traces = self._agent.query(message=user_input, session_id=session_id) - response = traces.get("response", "No response generated") - print(response) - - except KeyboardInterrupt: - print("\n\nAgent: Conversation interrupted. Goodbye!") - # Save events if EventLog exists - if hasattr(self._agent, "_event_log") and self._agent._event_log is not None: - self._agent._event_log.save(session_id) - # Save timeline (agent, codec, storage_config already set in __init__) - if hasattr(self._agent, "_timeline") and self._agent._timeline is not None: - self._agent._timeline.save(session_id) - break - except EOFError: - print("\n\nAgent: Input ended. Goodbye!") - # Save events if EventLog exists - if hasattr(self._agent, "_event_log") and self._agent._event_log is not None: - self._agent._event_log.save(session_id) - # Save timeline (agent, codec, storage_config already set in __init__) - if hasattr(self._agent, "_timeline") and self._agent._timeline is not None: - self._agent._timeline.save(session_id) - break - except Exception as e: - print(f"\nError: {e}") - print("Type '/help' for available commands or '/quit' to exit") - - # Save events if EventLog exists - if hasattr(self._agent, "_event_log") and self._agent._event_log is not None: - self._agent._event_log.save(session_id) - # Save timeline (agent, codec, storage_config already set in __init__) - if hasattr(self._agent, "_timeline") and self._agent._timeline is not None: - self._agent._timeline.save(session_id) + _run_coroutine_blocking(self.aconverse(initial_message=initial_message, session_id=session_id)) # ============================================================================ # ASYNC INTERACTIVE CONVERSATION INTERFACE @@ -434,3 +260,25 @@ async def _default_input_handler() -> str: # Save timeline (agent, codec, storage_config already set in __init__) if hasattr(self._agent, "_timeline") and self._agent._timeline is not None: self._agent._timeline.save(session_id) + + +def _run_coroutine_blocking(coro) -> None: + try: + asyncio.get_running_loop() + except RuntimeError: + asyncio.run(coro) + return + + result: dict[str, BaseException | None] = {"error": None} + + def _runner() -> None: + try: + asyncio.run(coro) + except BaseException as exc: + result["error"] = exc + + thread = threading.Thread(target=_runner) + thread.start() + thread.join() + if result["error"] is not None: + raise result["error"] diff --git a/tests/unit/core/test_communicator_converse_async.py b/tests/unit/core/test_communicator_converse_async.py new file mode 100644 index 0000000..b0df2c6 --- /dev/null +++ b/tests/unit/core/test_communicator_converse_async.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +import builtins +from typing import Any + +import pytest + +from dana.core.agent.components.communicator import Communicator + + +class _FakeAgent: + agent_type = "dana-librarian" + object_id = "dana-librarian" + available_agents: list[Any] = [] + available_resources: list[Any] = [] + available_workflows: list[Any] = [] + + def __init__(self) -> None: + self.aquery_calls: list[dict[str, Any]] = [] + self.query_called = False + + async def aquery(self, **kwargs: Any) -> dict[str, Any]: + self.aquery_calls.append(kwargs) + return {"response": "Rows: 1"} + + def query(self, **_kwargs: Any) -> dict[str, Any]: + self.query_called = True + raise AssertionError("Communicator.converse must use aquery()") + + +@pytest.mark.asyncio +async def test_converse_sync_wrapper_uses_async_agent_api_from_running_loop( + monkeypatch: pytest.MonkeyPatch, +) -> None: + agent = _FakeAgent() + inputs = iter(["/quit"]) + monkeypatch.setattr(builtins, "input", lambda _prompt="": next(inputs)) + + Communicator(agent).converse(initial_message="show rows", session_id="sess:one") + + assert agent.aquery_calls == [{"message": "show rows", "session_id": "sess:one"}] + assert not agent.query_called