andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-09-20 10:03:37 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

feat(tracing): collect human feedback and pause events in the trace (#7499)

Browse Source 
* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Vidit Ostwal <110953813+Vidit-Ostwal@users.noreply.github.com>
This commit is contained in:
João Moura
2026-09-16 13:41:09 -03:00
committed by GitHub
parent c0af9badb8
commit 32a9d2ae7b
2 changed files with 397 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py
View File

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

365
lib/crewai/tests/tracing/test_trace_human_feedback_events.py Normal file
View File

@@ -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) == []