fix: dispatch async retry stream events to async-only callbacks

The async retry loop emitted RETRY events via the sync emit() path, so
consumers registered through the public add_async_callback() API passed
has_callbacks but never received the event. Add an async-aware
_aemit_retry_stream_event helper (using emit_async) and share event
construction via _build_retry_stream_event. Adds tests for the async-only
callback path and the no-callback no-op case.

Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>

Author: praisonai-triage-agent[bot]

src/praisonai-agents/praisonaiagents/agent/chat_mixin.py

@@ -4432,19 +4432,51 @@ def _emit_retry_stream_event(self, *, attempt, max_attempts, delay, reason):
             return
         try:
             from ..streaming.events import StreamEvent, StreamEventType
-            emitter.emit(StreamEvent(
-                type=StreamEventType.RETRY,
-                metadata={
-                    "attempt": attempt,
-                    "max_attempts": max_attempts,
-                    "delay": delay,
-                    "reason": reason,
-                },
-                agent_id=getattr(self, 'name', None),
+            emitter.emit(self._build_retry_stream_event(
+                StreamEvent, StreamEventType,
+                attempt=attempt, max_attempts=max_attempts, delay=delay, reason=reason,
             ))
         except Exception as _re:
             logger.debug(f"Failed to emit RETRY stream event: {_re}")
 
+    async def _aemit_retry_stream_event(self, *, attempt, max_attempts, delay, reason):
+        """Async counterpart of ``_emit_retry_stream_event``.
+
+        Uses ``emit_async`` so consumers registered via the public
+        ``add_async_callback()`` API also receive RETRY events; ``emit`` alone
+        only visits synchronous callbacks. Same zero-overhead guard and
+        never-raise contract as the sync path.
+        """
+        emitter = getattr(self, 'stream_emitter', None)
+        if emitter is None or not getattr(emitter, 'has_callbacks', False):
+            return
+        try:
+            from ..streaming.events import StreamEvent, StreamEventType
+            event = self._build_retry_stream_event(
+                StreamEvent, StreamEventType,
+                attempt=attempt, max_attempts=max_attempts, delay=delay, reason=reason,
+            )
+            emit_async = getattr(emitter, 'emit_async', None)
+            if emit_async is not None:
+                await emit_async(event)
+            else:
+                emitter.emit(event)
+        except Exception as _re:
+            logger.debug(f"Failed to emit RETRY stream event: {_re}")
+
+    def _build_retry_stream_event(self, StreamEvent, StreamEventType, *, attempt, max_attempts, delay, reason):
+        """Construct the RETRY ``StreamEvent`` shared by the sync/async emitters."""
+        return StreamEvent(
+            type=StreamEventType.RETRY,
+            metadata={
+                "attempt": attempt,
+                "max_attempts": max_attempts,
+                "delay": delay,
+                "reason": reason,
+            },
+            agent_id=getattr(self, 'name', None),
+        )
+
     def _chat_completion_with_retry(self, messages, temperature=1.0, tools=None, stream=None, reasoning_steps=False, task_name=None, task_description=None, task_id=None, response_format=None, stream_callback=None, emit_events=True):
         """
         Wrapper for _execute_unified_chat_completion that adds jittered exponential backoff retry logic.
@@ -4508,20 +4540,14 @@ def _chat_completion_with_retry(self, messages, temperature=1.0, tools=None, str
 
                 # Surface the retry as a first-class stream event so CLI/UI
                 # consumers can show a live "retrying in Ns" status instead of
-                # appearing hung. Guarded by has_callbacks -> zero overhead when
-                # nothing is listening.
-                emitter = getattr(self, 'stream_emitter', None)
-                if emitter is not None and emitter.has_callbacks:
-                    from ..streaming.events import StreamEvent, StreamEventType
-                    emitter.emit(StreamEvent(
-                        type=StreamEventType.RETRY,
-                        metadata={
-                            "attempt": attempt + 1,
-                            "max_attempts": max_attempts,
-                            "delay": delay,
-                            "reason": str(e),
-                        },
-                    ))
+                # appearing hung. Guarded so it is zero-overhead when nothing
+                # is listening.
+                self._emit_retry_stream_event(
+                    attempt=attempt + 1,
+                    max_attempts=max_attempts,
+                    delay=delay,
+                    reason=str(e),
+                )
 
                 # Log retry attempt (buffered to avoid spam during transient failures)
                 logger.debug(f"[{self.name}] Retry {attempt + 1}/{max_attempts} after {delay:.1f}s: {str(e)[:100]}")
@@ -4604,20 +4630,15 @@ async def _achat_completion_with_retry(self, messages, temperature=1.0, tools=No
                 )
                 await self._hook_runner.execute_async(HookEvent.ON_RETRY, retry_input)
 
-                # Surface the retry as a first-class stream event (see sync path).
-                # Guarded by has_callbacks -> zero overhead when nothing is listening.
-                emitter = getattr(self, 'stream_emitter', None)
-                if emitter is not None and emitter.has_callbacks:
-                    from ..streaming.events import StreamEvent, StreamEventType
-                    await emitter.emit_async(StreamEvent(
-                        type=StreamEventType.RETRY,
-                        metadata={
-                            "attempt": attempt + 1,
-                            "max_attempts": max_attempts,
-                            "delay": delay,
-                            "reason": str(e),
-                        },
-                    ))
+                # Surface a streaming RETRY event (see sync path for rationale).
+                # Use the async emitter so async-only consumers registered via
+                # add_async_callback() also receive the event.
+                await self._aemit_retry_stream_event(
+                    attempt=attempt + 1,
+                    max_attempts=max_attempts,
+                    delay=delay,
+                    reason=str(e),
+                )
 
                 # Log retry attempt
                 logger.debug(f"[{self.name}] Async retry {attempt + 1}/{max_attempts} after {delay:.1f}s: {str(e)[:100]}")

src/praisonai-agents/tests/unit/test_streaming_events.py

@@ -360,14 +360,85 @@ def test_retry_event_carries_attempt_and_delay(self):
                 "attempt": 2,
                 "max_attempts": 4,
                 "delay": 8.0,
-                "reason": "rate limit",
+                "reason": "rate limited",
             },
         ))
 
         assert len(received) == 1
-        ev = received[0]
-        assert ev.type == StreamEventType.RETRY
-        assert ev.metadata["attempt"] == 2
-        assert ev.metadata["max_attempts"] == 4
-        assert ev.metadata["delay"] == 8.0
-        assert ev.metadata["reason"] == "rate limit"
+        evt = received[0]
+        assert evt.type == StreamEventType.RETRY
+        assert evt.metadata["attempt"] == 2
+        assert evt.metadata["max_attempts"] == 4
+        assert evt.metadata["delay"] == 8.0
+        assert evt.metadata["reason"] == "rate limited"
+
+    def test_emit_retry_stream_event_helper_zero_overhead_without_callbacks(self):
+        """The agent helper must be a no-op when no callbacks are attached."""
+        from praisonaiagents import Agent
+
+        agent = Agent(name="retry-agent")
+        # No callbacks attached -> helper should not raise and emit nothing.
+        received = []
+        agent.stream_emitter.add_callback(lambda e: received.append(e))
+        agent.stream_emitter.remove_callback  # sanity: emitter present
+        # Remove the callback to simulate the no-consumer case.
+        agent.stream_emitter._callbacks.clear()
+        assert agent.stream_emitter.has_callbacks is False
+        agent._emit_retry_stream_event(attempt=1, max_attempts=3, delay=1.0, reason="x")
+        assert received == []
+
+    def test_emit_retry_stream_event_helper_emits_when_listening(self):
+        """The agent helper should emit a RETRY event when a consumer listens."""
+        from praisonaiagents import Agent
+        from praisonaiagents.streaming.events import StreamEventType
+
+        agent = Agent(name="retry-agent")
+        received = []
+        agent.stream_emitter.add_callback(lambda e: received.append(e))
+        agent._emit_retry_stream_event(attempt=2, max_attempts=4, delay=5.0, reason="rate limited")
+
+        assert len(received) == 1
+        evt = received[0]
+        assert evt.type == StreamEventType.RETRY
+        assert evt.metadata == {
+            "attempt": 2,
+            "max_attempts": 4,
+            "delay": 5.0,
+            "reason": "rate limited",
+        }
+
+    def test_aemit_retry_reaches_async_only_callbacks(self):
+        """The async helper must reach consumers registered via add_async_callback()."""
+        import asyncio
+        from praisonaiagents import Agent
+        from praisonaiagents.streaming.events import StreamEventType
+
+        agent = Agent(name="retry-agent")
+        received = []
+
+        async def async_cb(event):
+            received.append(event)
+
+        # Async-only consumer: has_callbacks is True but sync emit() would miss it.
+        agent.stream_emitter.add_async_callback(async_cb)
+
+        asyncio.run(agent._aemit_retry_stream_event(
+            attempt=1, max_attempts=3, delay=2.0, reason="rate limited",
+        ))
+
+        assert len(received) == 1
+        assert received[0].type == StreamEventType.RETRY
+        assert received[0].metadata["attempt"] == 1
+        assert received[0].metadata["max_attempts"] == 3
+
+    def test_aemit_retry_zero_overhead_without_callbacks(self):
+        """The async helper must be a no-op when no callbacks are attached."""
+        import asyncio
+        from praisonaiagents import Agent
+
+        agent = Agent(name="retry-agent")
+        assert agent.stream_emitter.has_callbacks is False
+        # Should not raise and should emit nothing.
+        asyncio.run(agent._aemit_retry_stream_event(
+            attempt=1, max_attempts=3, delay=1.0, reason="x",
+        ))