fix: surface retry/backoff status in the CLI (fixes #3311)

Add StreamEventType.RETRY emitted alongside the existing ON_RETRY hook (guarded by has_callbacks for zero-overhead) in both sync and async retry loops, carrying attempt/max_attempts/delay/reason in metadata. Map it to a canonical run.retry NDJSON event in the CLI stream-json bridge so consumers get a live 'retrying in Ns (attempt k/N)' signal. Non-JSON/silent modes stay silent.

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

Author: praisonai-triage-agent[bot]

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

@@ -4419,6 +4419,32 @@ async def _apply_context_compaction_async(self, messages, hook_event_class):
             logging.debug(f"[compaction] skipped (non-fatal): {_ce}")
             return False
 
+    def _emit_retry_stream_event(self, *, attempt, max_attempts, delay, reason):
+        """Emit a ``StreamEventType.RETRY`` event during backoff.
+
+        Lets consumers (e.g. the CLI stream-json bridge) render a live
+        "retrying in Ns (attempt k/N)" status instead of appearing hung.
+        Guarded by ``has_callbacks`` so it stays zero-overhead when nothing is
+        listening, and never raises into the retry loop.
+        """
+        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
+            emitter.emit(StreamEvent(
+                type=StreamEventType.RETRY,
+                metadata={
+                    "attempt": attempt,
+                    "max_attempts": max_attempts,
+                    "delay": delay,
+                    "reason": reason,
+                },
+                agent_id=getattr(self, 'name', None),
+            ))
+        except Exception as _re:
+            logger.debug(f"Failed to emit RETRY stream event: {_re}")
+
     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.

src/praisonai/tests/unit/cli/test_run_event_bridge.py

@@ -27,13 +27,14 @@ def __init__(self, value):
 
 class _FakeStreamEvent:
     def __init__(self, type_value, content=None, tool_call=None,
-                 error=None, is_reasoning=False, agent_id=None):
+                 error=None, is_reasoning=False, agent_id=None, metadata=None):
         self.type = _FakeEnum(type_value)
         self.content = content
         self.tool_call = tool_call
         self.error = error
         self.is_reasoning = is_reasoning
         self.agent_id = agent_id
+        self.metadata = metadata
 
 
 class _FakeEmitter:
@@ -145,6 +146,18 @@ def test_retry_event_mapped_to_run_retry(capsys):
     assert retry["data"]["max_attempts"] == 4
     assert retry["data"]["delay"] == 8.0
     assert retry["data"]["reason"] == "rate limit"
+    assert retry["data"]["schema_version"] == SCHEMA_VERSION
+
+
+def test_retry_event_silent_in_text_mode(capsys):
+    output = OutputController(mode=OutputMode.TEXT)
+    agent = _FakeAgent()
+    bridge = attach_bridge(agent, output)
+    # Bridge is a no-op in non-JSON modes.
+    assert bridge is None
+    agent.stream_emitter.emit(_FakeStreamEvent(
+        "retry", metadata={"attempt": 1, "max_attempts": 3, "delay": 1.0}))
+    assert capsys.readouterr().out.strip() == ""
 
 
 def test_run_lifecycle_helpers(capsys):