From 5849f48ecef6364fceb615255cf2eb0fe7572bed Mon Sep 17 00:00:00 2001 From: Joao Moura Date: Tue, 11 Aug 2026 17:38:26 -0700 Subject: [PATCH] fix(telemetry): record the resumed marker as a string Verified end to end against the live collector and ClickHouse: the pipeline encodes a boolean attribute as the presence of a vBool key, so false arrives as the key simply being absent. That is invisible in the schema and easy to read wrongly - crew_memory is extracted as "the attribute exists" and consequently reports 1 for 99.8% of crews against a field that defaults to False. A string leaves nothing to infer. Confirmed in the warehouse: the emitted span reads resumed = "false". Adds direct coverage for the attributes each flow span records, including both resumed values, and resets the Telemetry singleton in the helper so more than one span method can be exercised per session. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01ASfWmW3RGy4qAQm6s8U9jH --- lib/crewai/src/crewai/telemetry/telemetry.py | 7 ++- lib/crewai/tests/telemetry/test_telemetry.py | 65 ++++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/lib/crewai/src/crewai/telemetry/telemetry.py b/lib/crewai/src/crewai/telemetry/telemetry.py index c96b827a49..3b7a2682e4 100644 --- a/lib/crewai/src/crewai/telemetry/telemetry.py +++ b/lib/crewai/src/crewai/telemetry/telemetry.py @@ -1024,7 +1024,12 @@ class Telemetry: self._add_attribute(span, "flow_name", flow_name) self._add_attribute(span, "node_names", json.dumps(node_names)) self._add_attribute(span, "origin", origin) - self._add_attribute(span, "resumed", resumed) + # Recorded as a string rather than a bool. The pipeline encodes a + # boolean as the presence of a vBool key - false arrives as the key + # simply being absent - which is invisible in the schema and easy to + # extract wrongly. crew_memory reads 1 for 99.8% of crews for exactly + # that reason, against a field that defaults to False. + self._add_attribute(span, "resumed", "true" if resumed else "false") close_span(span) self._safe_telemetry_operation(_operation) diff --git a/lib/crewai/tests/telemetry/test_telemetry.py b/lib/crewai/tests/telemetry/test_telemetry.py index acc44bf9ef..5c30bd4ffd 100644 --- a/lib/crewai/tests/telemetry/test_telemetry.py +++ b/lib/crewai/tests/telemetry/test_telemetry.py @@ -306,3 +306,68 @@ def test_event_listener_tracks_hook_dispatched_events(): interception_point="pre_tool_call", outcome="aborted", ) + + +def _emit(method: str, *args, **kwargs): + """Run one telemetry span method against a mocked tracer. + + The singleton is reset first: it caches the provider built on the very + first construction, so without this only the earliest caller in a session + would see the mocked tracer. + """ + tracer = Mock() + span = Mock() + tracer.start_span.return_value = span + Telemetry._instance = None + + with ( + patch.dict( + os.environ, + { + "CREWAI_DISABLE_TELEMETRY": "false", + "CREWAI_DISABLE_TRACKING": "false", + "OTEL_SDK_DISABLED": "false", + }, + ), + patch( + "crewai.telemetry.telemetry.TracerProvider", + return_value=Mock(get_tracer=Mock(return_value=tracer)), + ), + patch("crewai.telemetry.telemetry.version", return_value="9.9.9"), + ): + getattr(Telemetry(), method)(*args, **kwargs) + Telemetry._instance = None + return tracer, span + + +@pytest.mark.parametrize(("resumed", "expected"), [(True, "true"), (False, "false")]) +def test_resumed_is_recorded_as_a_string(resumed: bool, expected: str) -> None: + """A boolean is encoded as the presence of a key, not as a value. + + ``false`` arrives as the key simply being absent, which is invisible in the + schema and easy to extract wrongly - crew_memory reads 1 for 99.8% of crews + for exactly that reason. A string leaves nothing to infer. + """ + _tracer, span = _emit( + "flow_execution_span", "ResearchFlow", ["start"], "user", resumed + ) + + span.set_attribute.assert_any_call("resumed", expected) + for call in span.set_attribute.call_args_list: + assert call.args[1] is not True and call.args[1] is not False + + +def test_flow_completed_records_duration_outcome_and_origin() -> None: + _tracer, span = _emit("flow_completed_span", "ResearchFlow", 12.5, "failed", "user") + + span.set_attribute.assert_any_call("flow_name", "ResearchFlow") + span.set_attribute.assert_any_call("duration_ms", 12.5) + span.set_attribute.assert_any_call("outcome", "failed") + span.set_attribute.assert_any_call("origin", "user") + + +def test_paused_and_method_failed_record_flow_and_origin() -> None: + for method in ("flow_paused_span", "flow_method_failed_span"): + _tracer, span = _emit(method, "ResearchFlow", "internal") + span.set_attribute.assert_any_call("flow_name", "ResearchFlow") + span.set_attribute.assert_any_call("origin", "internal")