From 32a9d2ae7b44597d4ffbd110c57c405821708871 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Moura?= Date: Wed, 16 Sep 2026 13:41:09 -0300 Subject: [PATCH] feat(tracing): collect human feedback and pause events in the trace (#7499) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(tracing): collect human feedback and pause events in the trace The trace listener subscribed to method and conversation events but not to the review-gate events or the pause events, so a `@human_feedback` gate reached the trace only as method_execution_started/finished. A trace could not say that a run stopped for review, what the reviewer was shown, or what they answered. Subscribe to HumanFeedbackRequestedEvent, HumanFeedbackReceivedEvent, MethodExecutionPausedEvent and FlowPausedEvent through `_handle_action_event`, as the conversation events are, with the event's own type as the trace type. Each is a whole-event payload via the default serialization path; no change to `_build_event_data` or `complex_events`. Co-Authored-By: Claude Fable 5.1 * fix(tracing): register the gate and pause handlers through _on, keeping the execution-uuid gate The four new handlers, and the conversation handler this branch had switched by mistake, registered with event_bus.on and so ran while a kickoff owned an execution uuid — the case where the OTEL session records these events and the legacy collector must stay idle. Restored to self._on like every other handler; a test binds an execution uuid and asserts none of the five are collected into a legacy batch. Docstrings on the handlers (review bot coverage note). Co-Authored-By: Claude Fable 5.1 * test(tracing): under a tracing kickoff the session records the gate and pause events and the legacy batch stays empty Two real flows under an in-memory tracing session (the lifecycle tests' pattern): a @human_feedback gate answered at the console records human_feedback_requested and human_feedback_received as spans; an async provider that parks the flow records method_execution_paused and flow_paused. In both the legacy collector, gated by _on, collects none of the four. Review bot: the uuid-gated test alone would have passed with the session registrations missing. Co-Authored-By: Claude Fable 5.1 --------- Co-authored-by: Claude Fable 5.1 Co-authored-by: Vidit Ostwal <110953813+Vidit-Ostwal@users.noreply.github.com> --- .../listeners/tracing/trace_listener.py | 32 ++ .../test_trace_human_feedback_events.py | 365 ++++++++++++++++++ 2 files changed, 397 insertions(+) create mode 100644 lib/crewai/tests/tracing/test_trace_human_feedback_events.py diff --git a/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py b/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py index ba4a94313..ce3badb28 100644 --- a/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py +++ b/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py @@ -70,10 +70,14 @@ from crewai.events.types.flow_events import ( FlowCreatedEvent, FlowFailedEvent, FlowFinishedEvent, + FlowPausedEvent, FlowPlotEvent, FlowStartedEvent, + HumanFeedbackReceivedEvent, + HumanFeedbackRequestedEvent, MethodExecutionFailedEvent, MethodExecutionFinishedEvent, + MethodExecutionPausedEvent, MethodExecutionStartedEvent, ) from crewai.events.types.knowledge_events import ( @@ -280,6 +284,34 @@ class TraceCollectionListener(BaseEventListener): def on_method_failed(source: Any, event: MethodExecutionFailedEvent) -> None: self._handle_trace_event("method_execution_failed", source, event) + # Registered through `_on`, like every other handler here: while a kickoff + # owns an execution uuid the OTEL session records these events (see + # `telemetry/tracing/handlers.py`) and this legacy collector stays idle. + + @self._on(event_bus, MethodExecutionPausedEvent) + def on_method_paused(source: Any, event: MethodExecutionPausedEvent) -> None: + """Collect a method pausing for human feedback, whole.""" + self._handle_action_event("method_execution_paused", source, event) + + @self._on(event_bus, HumanFeedbackRequestedEvent) + def on_human_feedback_requested( + source: Any, event: HumanFeedbackRequestedEvent + ) -> None: + """Collect what the reviewer was shown when a gate asked for feedback.""" + self._handle_action_event("human_feedback_requested", source, event) + + @self._on(event_bus, HumanFeedbackReceivedEvent) + def on_human_feedback_received( + source: Any, event: HumanFeedbackReceivedEvent + ) -> None: + """Collect the reviewer's answer and where the flow routed on it.""" + self._handle_action_event("human_feedback_received", source, event) + + @self._on(event_bus, FlowPausedEvent) + def on_flow_paused(source: Any, event: FlowPausedEvent) -> None: + """Collect a flow pausing (a gate or an explicit pause), whole.""" + self._handle_action_event("flow_paused", source, event) + @self._on(event_bus, ConversationMessageAddedEvent) def on_conversation_message_added( source: Any, event: ConversationMessageAddedEvent diff --git a/lib/crewai/tests/tracing/test_trace_human_feedback_events.py b/lib/crewai/tests/tracing/test_trace_human_feedback_events.py new file mode 100644 index 000000000..bb7e8a78d --- /dev/null +++ b/lib/crewai/tests/tracing/test_trace_human_feedback_events.py @@ -0,0 +1,365 @@ +"""A `@human_feedback` gate must reach the trace as more than method start/finish. + +The listener subscribed to method and conversation events but not to the +review-gate events (`HumanFeedbackRequestedEvent`, `HumanFeedbackReceivedEvent`) +or the pause events (`MethodExecutionPausedEvent`, `FlowPausedEvent`). A trace +could therefore not say that a run stopped for review, what the reviewer was +shown, or what they answered. Each is collected as a whole-event payload, like +every other non-complex type. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from datetime import datetime, timedelta, timezone +import os +from typing import Any +from unittest.mock import patch + +from crewai.events.event_bus import crewai_event_bus +from crewai.events.listeners.tracing.trace_listener import TraceCollectionListener +from crewai.execution import clear_execution_uuid, set_execution_uuid +from crewai.flow.async_feedback import HumanFeedbackPending, PendingFeedbackContext +from crewai.flow.flow import Flow, listen, start +from crewai.flow.human_feedback import human_feedback +from crewai.flow.persistence.base import FlowPersistence +from crewai.telemetry.tracing.grants import GrantSpanExporter, TraceGrant, TraceGrantClient +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from crewai.events.listeners.tracing.types import TraceEvent +from crewai.events.types.flow_events import ( + FlowPausedEvent, + HumanFeedbackReceivedEvent, + HumanFeedbackRequestedEvent, + MethodExecutionPausedEvent, +) +import pytest + + +@pytest.fixture +def listener() -> Iterator[TraceCollectionListener]: + """A listener with tracing enabled and every handler registered. + + Enabled through the environment like the other tracing tests. Only the + env var is set, not the tracing context var, so batch initialization + never reaches the backend: `_initialize_backend_batch` returns before + any request when tracing is not enabled in context. + + `scoped_handlers` is required: the bus is a process-wide singleton, so + handlers registered here would otherwise fire in whatever runs next. + """ + with ( + crewai_event_bus.scoped_handlers(), + patch.dict(os.environ, {"CREWAI_TRACING_ENABLED": "true"}), + ): + trace_listener = TraceCollectionListener() + trace_listener.setup_listeners(crewai_event_bus) + yield trace_listener + + +def _collected(listener: TraceCollectionListener, event_type: str) -> TraceEvent: + """The single collected trace event of `event_type`.""" + crewai_event_bus.flush() + matches = [e for e in listener.batch_manager.event_buffer if e.type == event_type] + assert len(matches) == 1, ( + f"expected one {event_type!r} trace event, got {len(matches)}: " + f"{[e.type for e in listener.batch_manager.event_buffer]}" + ) + return matches[0] + + +def test_human_feedback_requested_is_collected_whole(listener) -> None: + crewai_event_bus.emit( + None, + HumanFeedbackRequestedEvent( + flow_name="review_flow", + method_name="draft", + output="the draft shown to the reviewer", + message="Approve this draft?", + emit=["approved", "rejected"], + request_id="req-1", + ), + ) + + data = _collected(listener, "human_feedback_requested").event_data + + assert data["flow_name"] == "review_flow" + assert data["method_name"] == "draft" + assert data["output"] == "the draft shown to the reviewer" + assert data["message"] == "Approve this draft?" + assert data["emit"] == ["approved", "rejected"] + assert data["request_id"] == "req-1" + + +def test_human_feedback_received_is_collected_whole(listener) -> None: + crewai_event_bus.emit( + None, + HumanFeedbackReceivedEvent( + flow_name="review_flow", + method_name="draft", + feedback="Looks good, ship it.", + outcome="approved", + request_id="req-1", + ), + ) + + data = _collected(listener, "human_feedback_received").event_data + + assert data["flow_name"] == "review_flow" + assert data["method_name"] == "draft" + assert data["feedback"] == "Looks good, ship it." + assert data["outcome"] == "approved" + assert data["request_id"] == "req-1" + + +def test_method_execution_paused_is_collected_whole(listener) -> None: + crewai_event_bus.emit( + None, + MethodExecutionPausedEvent( + flow_name="review_flow", + method_name="draft", + state={"draft": "v1"}, + flow_id="flow-1", + message="Approve this draft?", + emit=["approved", "rejected"], + ), + ) + + data = _collected(listener, "method_execution_paused").event_data + + assert data["flow_name"] == "review_flow" + assert data["method_name"] == "draft" + assert data["flow_id"] == "flow-1" + assert data["state"] == {"draft": "v1"} + assert data["message"] == "Approve this draft?" + assert data["emit"] == ["approved", "rejected"] + + +def test_flow_paused_is_collected_whole(listener) -> None: + crewai_event_bus.emit( + None, + FlowPausedEvent( + flow_name="review_flow", + flow_id="flow-1", + method_name="draft", + state={"draft": "v1"}, + message="Approve this draft?", + emit=["approved", "rejected"], + ), + ) + + data = _collected(listener, "flow_paused").event_data + + assert data["flow_name"] == "review_flow" + assert data["flow_id"] == "flow-1" + assert data["method_name"] == "draft" + assert data["state"] == {"draft": "v1"} + assert data["message"] == "Approve this draft?" + assert data["emit"] == ["approved", "rejected"] + + +def test_the_new_handlers_stay_idle_while_a_kickoff_owns_an_execution_uuid(listener) -> None: + """The legacy collector must not run beside the OTEL session. + + Every handler in the listener registers through `_on`, which skips the + event when an execution uuid is bound (the kickoff owns the new session and + `telemetry/tracing/handlers.py` records these events there). The four gate + and pause handlers, and the conversation handler, keep that gate: a run + under an execution uuid collects none of them into a legacy batch. + """ + token = set_execution_uuid("exec-owned-by-the-otel-session") + try: + crewai_event_bus.emit( + object(), + HumanFeedbackRequestedEvent( + flow_name="review_flow", + method_name="draft", + output="the draft shown to the reviewer", + message="Approve this draft?", + emit=["approved", "rejected"], + request_id="req-2", + ), + ) + crewai_event_bus.emit( + object(), + HumanFeedbackReceivedEvent( + flow_name="review_flow", + method_name="draft", + feedback="Looks good, ship it.", + outcome="approved", + request_id="req-2", + ), + ) + crewai_event_bus.emit( + object(), + MethodExecutionPausedEvent( + flow_name="review_flow", + method_name="draft", + state={"draft": "v1"}, + flow_id="flow-2", + message="Approve this draft?", + emit=["approved", "rejected"], + ), + ) + crewai_event_bus.emit( + object(), + FlowPausedEvent( + flow_name="review_flow", + flow_id="flow-2", + method_name="draft", + state={"draft": "v1"}, + message="Approve this draft?", + emit=["approved", "rejected"], + ), + ) + crewai_event_bus.flush() + finally: + clear_execution_uuid(token) + + collected = [e.type for e in listener.batch_manager.event_buffer] + assert not any( + t in collected + for t in ( + "human_feedback_requested", + "human_feedback_received", + "method_execution_paused", + "flow_paused", + ) + ), collected + + +# --------------------------------------------------------------------------- +# Under a tracing kickoff the OTEL session records the four events and the +# legacy collector stays idle — the two halves of the gate, on real flows. +# --------------------------------------------------------------------------- + +LEGACY_TYPES = ( + "human_feedback_requested", + "human_feedback_received", + "method_execution_paused", + "flow_paused", +) + + +@pytest.fixture +def session_recorders(monkeypatch) -> dict[str, InMemorySpanExporter]: + """A tracing session that exports to memory, as in tests/telemetry/test_trace_lifecycle.py. + + No socket is opened: the grant is synthetic and the exporter records + spans per execution uuid; anonymous consent is granted without a prompt. + """ + monkeypatch.delenv("OTEL_SDK_DISABLED", raising=False) + monkeypatch.delenv("CREWAI_USER_PAT", raising=False) + monkeypatch.delenv("CREWAI_PLATFORM_INTEGRATION_TOKEN", raising=False) + monkeypatch.setenv("CREWAI_TRACING_ENABLED", "true") + monkeypatch.setenv("CREWAI_DISABLE_TELEMETRY", "true") + monkeypatch.setattr("crewai.telemetry.tracing.grants.get_auth_token", lambda: None) + recorders: dict[str, InMemorySpanExporter] = {} + + def create(client: Any, execution_uuid: str) -> TraceGrant: + return TraceGrant( + token="synthetic-grant", + collector_url="https://collector.invalid/v1/traces", + execution_uuid=execution_uuid, + expires_at=datetime.now(timezone.utc) + timedelta(minutes=15), + ) + + def exporter(grant: TraceGrant) -> InMemorySpanExporter: + recorder = InMemorySpanExporter() + recorders[grant.execution_uuid] = recorder + return recorder + + monkeypatch.setattr(TraceGrantClient, "create", create) + monkeypatch.setattr(GrantSpanExporter, "_exporter", staticmethod(exporter)) + return recorders + + +def _session_event_names(recorders: dict[str, InMemorySpanExporter]) -> set[str]: + """Every `crewai.event_name` the session recorded, across executions.""" + return { + str(span.attributes.get("crewai.event_name")) + for recorder in recorders.values() + for span in recorder.get_finished_spans() + if span.attributes and span.attributes.get("crewai.event_name") + } + + +def _legacy_types(listener: TraceCollectionListener) -> list[str]: + crewai_event_bus.flush() + return [e.type for e in listener.batch_manager.event_buffer if e.type in LEGACY_TYPES] + + +def test_a_gate_answered_in_place_is_recorded_by_the_session_not_the_legacy_batch( + listener, session_recorders +) -> None: + """A `@human_feedback` gate answered at the console, under a tracing kickoff. + + The kickoff owns an execution uuid, so the OTEL session records + `human_feedback_requested` and `human_feedback_received` as spans and the + legacy collector — gated by `_on` — collects neither. + """ + + class ReviewFlow(Flow): + @start() + @human_feedback(message="Approve this draft?") + def draft(self) -> str: + return "the draft" + + @listen(draft) + def finish(self, result) -> str: + return f"done: {result.feedback}" + + with ( + patch("builtins.input", return_value="looks good"), + patch( + "crewai.telemetry.tracing.ephemeral.prompt_user_for_trace_viewing", + return_value=True, + ), + ): + result = ReviewFlow(tracing=True).kickoff() + + assert result == "done: looks good" + recorded = _session_event_names(session_recorders) + assert {"human_feedback_requested", "human_feedback_received"} <= recorded, recorded + assert _legacy_types(listener) == [] + + +def test_a_gate_that_pauses_the_flow_is_recorded_by_the_session_not_the_legacy_batch( + listener, session_recorders +) -> None: + """An async provider parks the flow: the session records the two pause + events; the legacy collector, gated by `_on`, collects neither.""" + + class MemoryPersistence(FlowPersistence): + def init_db(self) -> None: + pass + + def save_state(self, flow_uuid, method_name, state_data) -> None: + pass + + def load_state(self, flow_uuid): + return None + + class AsyncProvider: + def request_feedback(self, context: PendingFeedbackContext, flow: Flow) -> str: + raise HumanFeedbackPending(context=context) + + class PausingFlow(Flow): + @start() + @human_feedback(message="Approve this draft?", provider=AsyncProvider()) + def draft(self) -> str: + return "the draft" + + @listen(draft) + def finish(self, result) -> str: + return f"done: {result.feedback}" + + with patch( + "crewai.telemetry.tracing.ephemeral.prompt_user_for_trace_viewing", + return_value=True, + ): + result = PausingFlow(persistence=MemoryPersistence(), tracing=True).kickoff() + + assert isinstance(result, HumanFeedbackPending) + recorded = _session_event_names(session_recorders) + assert {"method_execution_paused", "flow_paused"} <= recorded, recorded + assert _legacy_types(listener) == []