fix: unblock STAR REPL under asyncio — recursion-safe __repr__, loud sync guard, async converse

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))
 

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

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"]

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:
         """

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