diff --git a/dana/common/llm/llm.py b/dana/common/llm/llm.py index 56ed792..31d0a65 100644 --- a/dana/common/llm/llm.py +++ b/dana/common/llm/llm.py @@ -295,18 +295,28 @@ 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 + # 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() - # 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 + 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." + ) - # 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)) 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/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/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/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 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..92bfe34 --- /dev/null +++ b/tests/unit/test_llm_chat_response_sync_bridge.py @@ -0,0 +1,73 @@ +"""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 + +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 (no running loop) uses run_until_complete.""" + 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_raises_loud_inside_running_loop(): + """Sync API called from within a running loop must raise a clear, actionable error. + + Previously: the dead guard let it fall through to run_until_complete → + "Cannot run the event loop while another loop is running" (cryptic). + """ + llm = _llm_returning(LLMResponse(content="ok", model="m")) + + async def caller(): + return llm.chat_response_sync(_user_msg()) + + 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_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")) + + with pytest.raises(RuntimeError, match="boom"): + llm.chat_response_sync(_user_msg()) 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