feat(replay): complete run events (run/parent ids, tasks, steps, hand-offs, approvals, LiteLLM) and graph export for Replay

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

@@ -630,6 +630,27 @@ async def _execute_with_runtime_resolution_async(
                 None, lambda: self.agent.chat(prompt, tools=effective_tools)
             )
 
+    def _trace_handoff(self, source_agent: 'Agent', context: Optional[Dict[str, Any]], reason: Optional[str] = None) -> None:
+        """Record the hand-off on the replay trace (no-op when tracing is off)."""
+        from ..trace.context_events import get_context_emitter
+        emitter = get_context_emitter()
+        if not emitter._enabled:
+            return
+        context = dict(context or {})
+        if reason is None:
+            reason = context.get("reason") or context.get("task") or context.get("prompt")
+        policy = getattr(self.config, "context_policy", None)
+        emitter.agent_handoff(
+            getattr(source_agent, "name", None) or "agent",
+            getattr(self.agent, "name", None),
+            reason=emitter._preview(reason) if reason is not None else None,
+            context_passed={
+                "context_policy": getattr(policy, "value", policy),
+                "args": context,
+            },
+            kind="handoff",
+        )
+
     def _check_safety(self, source_agent: 'Agent') -> None:
         """
         Check safety constraints before handoff.
@@ -778,6 +799,7 @@ def execute_programmatic(
             # Track handoff chain
             _push_handoff(source_agent.name)
             pushed = True
+            self._trace_handoff(source_agent, kwargs, reason=prompt)
 
             # Execute on_handoff callback
             self._execute_callback(self.config.on_handoff or self.on_handoff, source_agent, kwargs)
@@ -888,6 +910,7 @@ async def _execute():
                 self._check_safety(source_agent)
                 _push_handoff(source_agent.name)
                 pushed = True
+                self._trace_handoff(source_agent, kwargs, reason=prompt)
 
                 # Execute callback
                 self._execute_callback(self.config.on_handoff or self.on_handoff, source_agent, kwargs)
@@ -1016,6 +1039,7 @@ def handoff_tool(**kwargs):
                 # Track handoff chain
                 _push_handoff(source_agent.name)
                 pushed = True
+                self._trace_handoff(source_agent, kwargs, reason=None)
 
                 # Execute on_handoff callback
                 self._execute_callback(self.config.on_handoff or self.on_handoff, source_agent, kwargs)
@@ -1632,6 +1656,7 @@ def execute_programmatic(
             # Track handoff chain
             _push_handoff(source_agent.name)
             pushed = True
+            self._trace_handoff(source_agent, kwargs, reason=None)
 
             # Execute on_handoff callback
             self._execute_callback(self.config.on_handoff or self.on_handoff, source_agent, kwargs)
@@ -1747,6 +1772,7 @@ async def _execute():
                 self._check_safety(source_agent)
                 _push_handoff(source_agent.name)
                 pushed = True
+                self._trace_handoff(source_agent, kwargs, reason=None)
 
                 # Execute callback
                 self._execute_callback(self.config.on_handoff or self.on_handoff, source_agent, kwargs)

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

@@ -356,6 +356,39 @@ def _record_skipped_task(agents_instance, task):
     task.status = "completed"
 
 
+def _trace_task_begin(agents_instance, task):
+    """Emit ``task_start``; returns a span for ``_trace_task_end`` (None when off)."""
+    from ..trace.context_events import get_context_emitter
+    emitter = get_context_emitter()
+    if not emitter._enabled:
+        return None
+    from ..workflows.diagram import task_node_id
+    node_id = task_node_id(agents_instance.tasks, task)
+    agent = getattr(getattr(task, "agent", None), "name", None)
+    emitter.task_start(
+        getattr(task, "name", None), getattr(task, "description", None),
+        agent=agent, node_id=node_id,
+    )
+    return emitter, node_id, agent, time.perf_counter()
+
+
+def _trace_task_end(span, task, status):
+    """Emit ``task_end`` (completed / failed / skipped) for a ``_trace_task_begin`` span."""
+    if span is None:
+        return
+    emitter, node_id, agent, started = span
+    result = getattr(task, "result", None)
+    emitter.task_end(
+        getattr(task, "name", None),
+        status,
+        retries=getattr(task, "retry_count", 0) or 0,
+        duration_ms=(time.perf_counter() - started) * 1000,
+        output=getattr(result, "raw", None) if result is not None else None,
+        agent=agent,
+        node_id=node_id,
+    )
+
+
 def _task_should_skip(agents_instance, task):
     """Return True when a task's should_run gate evaluates falsy.
 
@@ -1539,7 +1572,19 @@ def _apply_human_review(self, task, task_id, task_output):
             agent_name=getattr(getattr(task, "agent", None), "name", None),
             context={"kind": "task_output_review", "prompt": prompt},
         )
-        decision = backend.request_approval_sync(request)
+        from ..trace.context_events import _approval_begin, _approval_end, get_context_emitter
+        _approval = None
+        if get_context_emitter()._enabled:
+            from ..workflows.diagram import task_node_id
+            _approval = _approval_begin(
+                request, kind="task_output_review", node_id=task_node_id(self.tasks, task),
+            )
+        try:
+            decision = backend.request_approval_sync(request)
+        except BaseException as exc:
+            _approval_end(_approval, error=exc)
+            raise
+        _approval_end(_approval, decision)
 
         if getattr(decision, "approved", False):
             return task_output, False
@@ -1740,14 +1785,26 @@ async def arun_task(self, task_id):
             logger.info(f"Task with ID {task_id} is already completed")
             return
 
+        _span = _trace_task_begin(self, task)
+        _status = "failed"
+        try:
+            if await self._arun_task_body(task_id, task) == "skipped":
+                _status = "skipped"
+            elif task.status == "completed":
+                _status = "completed"
+        finally:
+            _trace_task_end(_span, task, _status)
+
+    async def _arun_task_body(self, task_id, task):
+        """Body of :meth:`arun_task`; returns ``"skipped"`` for a gated-out task."""
         # Call on_task_start callback and propagate variables (async-aware, mirrors run_task)
         await self._arun_task_start_hook(task, task_id)
 
         # Honour the should_run conditional gate (mirrors the Workflow engine).
         if await _atask_should_skip(self, task):
             logger.info(f"Task {task_id} skipped by should_run gate")
             _record_skipped_task(self, task)
-            return
+            return "skipped"
 
         # Use per-task max_retries if available
         task_max = getattr(task, "max_retries", self.max_retries)
@@ -2201,7 +2258,12 @@ async def astart(self, content=None, return_dict=False, **kwargs):
         # Re-entrant on the owning thread so batch runs (astart_for_each) can hold
         # the guard across the whole batch while still calling astart() per item.
         with self._guard_execution():
-            return await self._astart_impl(content=content, return_dict=return_dict, **kwargs)
+            from ..trace.context_events import _workflow_scope
+            with _workflow_scope("team", self._replay_name(), self.process,
+                                 graph_fn=self.to_graph, run_id=self.run_id) as _run:
+                result = await self._astart_impl(content=content, return_dict=return_dict, **kwargs)
+                self._mark_run_status(_run)
+                return result
 
     async def _astart_impl(self, content=None, return_dict=False, **kwargs):
         # Track execution via telemetry
@@ -2318,14 +2380,26 @@ def run_task(self, task_id):
             logger.info(f"Task with ID {task_id} is already completed")
             return
 
+        _span = _trace_task_begin(self, task)
+        _status = "failed"
+        try:
+            if self._run_task_body(task_id, task) == "skipped":
+                _status = "skipped"
+            elif task.status == "completed":
+                _status = "completed"
+        finally:
+            _trace_task_end(_span, task, _status)
+
+    def _run_task_body(self, task_id, task):
+        """Body of :meth:`run_task`; returns ``"skipped"`` for a gated-out task."""
         # Call on_task_start callback and propagate variables (shared with arun_task)
         self._run_task_start_hook(task, task_id)
 
         # Honour the should_run conditional gate (mirrors the Workflow engine).
         if _task_should_skip(self, task):
             logger.info(f"Task {task_id} skipped by should_run gate")
             _record_skipped_task(self, task)
-            return
+            return "skipped"
 
         # Use per-task max_retries if available
         task_max = getattr(task, "max_retries", self.max_retries)
@@ -2441,6 +2515,21 @@ def run_all_tasks(self):
             except StopIteration:
                 pass
 
+    def to_graph(self) -> Dict[str, Any]:
+        """This team's definition as graph data (``praisonai.graph/1``).
+
+        Covers agents, tasks (context/depends_on, next_tasks, conditions),
+        tools, hand-offs, human review and the hierarchical manager. Node ids
+        are stable across processes and match replay events' ``node_id``.
+        """
+        from ..workflows.diagram import to_graph
+        return to_graph(self)
+
+    def to_mermaid(self) -> str:
+        """Render this team's definition as a mermaid diagram."""
+        from ..workflows.diagram import graph_to_mermaid, to_graph
+        return graph_to_mermaid(to_graph(self))
+
     def get_task_status(self, task_id):
         if task_id in self.tasks:
             return self.tasks[task_id].status
@@ -2519,8 +2608,27 @@ def start(self, content=None, return_dict=False, output=None, **kwargs):
         # Re-entrant on the owning thread so batch runs (start_for_each) can hold
         # the guard across the whole batch while still calling start() per item.
         with self._guard_execution():
-            return self._start_impl(content=content, return_dict=return_dict,
-                                    output=output, **kwargs)
+            from ..trace.context_events import _workflow_scope
+            with _workflow_scope("team", self._replay_name(), self.process,
+                                 graph_fn=self.to_graph, run_id=self.run_id) as _run:
+                result = self._start_impl(content=content, return_dict=return_dict,
+                                          output=output, **kwargs)
+                self._mark_run_status(_run)
+                return result
+
+    def _replay_name(self) -> str:
+        """Run label for Replay: the team's name, else its agents joined ("A + B")."""
+        if self.name:
+            return self.name
+        names = [getattr(a, "name", None) or "Agent" for a in (self.agents or [])]
+        return " + ".join(names) if names else "Agent team"
+
+    def _mark_run_status(self, run_state):
+        """Record a failed replay run when any task ended failed (tracing on only)."""
+        if run_state is not None and any(
+            getattr(t, "status", None) == "failed" for t in self.tasks.values()
+        ):
+            run_state["status"] = "failed"
 
     def _start_impl(self, content=None, return_dict=False, output=None, **kwargs):
         # Track execution via telemetry

src/praisonai-agents/praisonaiagents/approval/registry.py

@@ -689,29 +689,41 @@ def approve_sync(
             liveness=liveness,
         )
 
-        with _human_wait_scope():
-            # Prefer sync method if available
-            if hasattr(backend, "request_approval_sync"):
-                decision = backend.request_approval_sync(request)
-            else:
-                # Use shared utility for consistent async-to-sync bridging
-                from .utils import run_coroutine_safely
-                decision = run_coroutine_safely(
-                    backend.request_approval(request),
-                    timeout=self.timeout
-                )
-
-        # Live-authority gate: a resolution that lands after the originating
-        # turn was stopped/superseded is dropped fail-closed — the tool does not
-        # run and no durable session/always grant is persisted.
-        stale = self._reject_if_stale(request)
-        if stale is not None:
-            return stale
-
-        if decision.approved:
-            self.mark_approved(tool_name, arguments, agent_name, scope_id)
-        self._persist_scoped_decision(agent_name, tool_name, arguments, decision, scope_id)
-        return decision
+        # Replay: a person is actually being asked (the fast paths above are not).
+        from ..trace.context_events import _approval_begin, _approval_end
+        _trace = _approval_begin(request)
+        try:
+            with _human_wait_scope():
+                # Prefer sync method if available
+                if hasattr(backend, "request_approval_sync"):
+                    decision = backend.request_approval_sync(request)
+                else:
+                    # Use shared utility for consistent async-to-sync bridging
+                    from .utils import run_coroutine_safely
+                    decision = run_coroutine_safely(
+                        backend.request_approval(request),
+                        timeout=self.timeout
+                    )
+
+            # Live-authority gate: a resolution that lands after the originating
+            # turn was stopped/superseded is dropped fail-closed — the tool does not
+            # run and no durable session/always grant is persisted.
+            stale = self._reject_if_stale(request)
+            if stale is not None:
+                decision = stale
+                return stale
+
+            if decision.approved:
+                self.mark_approved(tool_name, arguments, agent_name, scope_id)
+            self._persist_scoped_decision(agent_name, tool_name, arguments, decision, scope_id)
+            return decision
+        except BaseException as exc:
+            _approval_end(_trace, error=exc)
+            _trace = None
+            raise
+        finally:
+            if _trace is not None:
+                _approval_end(_trace, decision)
 
     async def approve_async(
         self,
@@ -764,26 +776,38 @@ async def approve_async(
             liveness=liveness,
         )
 
+        # Replay: a person is actually being asked (the fast paths above are not).
+        from ..trace.context_events import _approval_begin, _approval_end
+        _trace = _approval_begin(request)
         try:
-            async with _human_wait_scope():
-                decision = await asyncio.wait_for(
-                    backend.request_approval(request),
-                    timeout=self.timeout,
-                )
-        except asyncio.TimeoutError:
-            decision = ApprovalDecision(approved=False, reason="Approval timed out")
-
-        # Live-authority gate: a resolution that lands after the originating
-        # turn was stopped/superseded is dropped fail-closed — the tool does not
-        # run and no durable session/always grant is persisted.
-        stale = self._reject_if_stale(request)
-        if stale is not None:
-            return stale
-
-        if decision.approved:
-            self.mark_approved(tool_name, arguments, agent_name, scope_id)
-        self._persist_scoped_decision(agent_name, tool_name, arguments, decision, scope_id)
-        return decision
+            try:
+                async with _human_wait_scope():
+                    decision = await asyncio.wait_for(
+                        backend.request_approval(request),
+                        timeout=self.timeout,
+                    )
+            except asyncio.TimeoutError:
+                decision = ApprovalDecision(approved=False, reason="Approval timed out")
+
+            # Live-authority gate: a resolution that lands after the originating
+            # turn was stopped/superseded is dropped fail-closed — the tool does not
+            # run and no durable session/always grant is persisted.
+            stale = self._reject_if_stale(request)
+            if stale is not None:
+                decision = stale
+                return stale
+
+            if decision.approved:
+                self.mark_approved(tool_name, arguments, agent_name, scope_id)
+            self._persist_scoped_decision(agent_name, tool_name, arguments, decision, scope_id)
+            return decision
+        except BaseException as exc:
+            _approval_end(_trace, error=exc)
+            _trace = None
+            raise
+        finally:
+            if _trace is not None:
+                _approval_end(_trace, decision)
 
     @staticmethod
     def _reject_if_stale(request: ApprovalRequest) -> Optional[ApprovalDecision]: