From d74e647502f5449c03873ae7ecaeba3b5d6d8368 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Moura?= Date: Thu, 13 Aug 2026 20:57:37 -0300 Subject: [PATCH] fix(core): record the running release on every emitted span (#6989) * fix(core): record the running release on every emitted span Nine of twenty-four span kinds never recorded crewai_version, including the two highest-volume ones - Task Created and Task Execution - plus Human Feedback, Flow Plotting, and the whole deployment family. add_crew_attributes writes crew_key, crew_id and crew_fingerprint but never the release, so any question filtered by version silently returned nothing for those spans and per-release comparison was blind to them. Add it at the fourteen sites that were missing it across both emitters, matching each module's existing convention: version("crewai") in crewai, get_crewai_version() with the file's local-import pattern in crewai_core. Guarded by a test that parses both modules and fails when any method creates a span without recording the release, so a span added later cannot reintroduce the gap. Verified non-vacuous: removing the attribute from one span makes it fail and names that method. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01ASfWmW3RGy4qAQm6s8U9jH * test(core): count spans against release attributes, and cover both emitters Two review findings, both real. The guard only asked whether a method mentioned crewai_version anywhere, so a method opening two spans while recording the release on one of them passed. task_started is exactly that shape. It now counts start_span calls against _add_attribute(..., "crewai_version", ...) calls and fails when the second is smaller, naming the method and both counts. Verified non-vacuous: removing the attribute from Task Execution alone - which the previous version accepted - now fails with "task_started (2 span(s), 1 version attribute(s))". The behavioural cases only ever ran against crewai's emitter, because _emit builds that singleton, so the five changed crewai_core methods had no behavioural coverage at all. Added a parametrized case over all eight spans crewai_core emits, using the fixture already in that file - covering the three that already recorded the release as well, so a regression there is caught too. Also removed the function-local `import crewai`: the paths now come from inspect.getfile() on the two classes, which is both consistent with the file's existing import style and more direct than guessing the module layout. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01ASfWmW3RGy4qAQm6s8U9jH --------- Co-authored-by: Claude Opus 5 (1M context) --- lib/crewai-core/src/crewai_core/telemetry.py | 15 ++ .../tests/test_telemetry_deploy.py | 43 ++++++ lib/crewai/src/crewai/telemetry/telemetry.py | 9 ++ lib/crewai/tests/telemetry/test_telemetry.py | 131 +++++++++++++++++- 4 files changed, 197 insertions(+), 1 deletion(-) diff --git a/lib/crewai-core/src/crewai_core/telemetry.py b/lib/crewai-core/src/crewai_core/telemetry.py index a3c2693d3..a7326f43b 100644 --- a/lib/crewai-core/src/crewai_core/telemetry.py +++ b/lib/crewai-core/src/crewai_core/telemetry.py @@ -293,9 +293,12 @@ class Telemetry: def deploy_signup_error_span(self) -> None: """Records when an error occurs during the deployment signup process.""" + from crewai_core.version import get_crewai_version + def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Deploy Signup Error") + self._add_attribute(span, "crewai_version", get_crewai_version()) close_span(span) self._safe_telemetry_procedure(_operation) @@ -313,9 +316,12 @@ class Telemetry: source: Where the deployment was initiated from. """ + from crewai_core.version import get_crewai_version + def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Start Deployment") + self._add_attribute(span, "crewai_version", get_crewai_version()) if uuid: self._add_attribute(span, "uuid", uuid) self._add_attribute(span, "source", source) @@ -334,9 +340,12 @@ class Telemetry: source: Where the deployment was initiated from. """ + from crewai_core.version import get_crewai_version + def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Create Crew Deployment") + self._add_attribute(span, "crewai_version", get_crewai_version()) self._add_attribute(span, "source", source) close_span(span) @@ -348,9 +357,12 @@ class Telemetry: ) -> None: """Records the retrieval of crew logs.""" + from crewai_core.version import get_crewai_version + def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Get Crew Logs") + self._add_attribute(span, "crewai_version", get_crewai_version()) self._add_attribute(span, "log_type", log_type) if uuid: self._add_attribute(span, "uuid", uuid) @@ -361,9 +373,12 @@ class Telemetry: def remove_crew_span(self, uuid: str | None = None) -> None: """Records the removal of a crew.""" + from crewai_core.version import get_crewai_version + def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Remove Crew") + self._add_attribute(span, "crewai_version", get_crewai_version()) if uuid: self._add_attribute(span, "uuid", uuid) close_span(span) diff --git a/lib/crewai-core/tests/test_telemetry_deploy.py b/lib/crewai-core/tests/test_telemetry_deploy.py index 0942f4654..b8ed653e2 100644 --- a/lib/crewai-core/tests/test_telemetry_deploy.py +++ b/lib/crewai-core/tests/test_telemetry_deploy.py @@ -136,3 +136,46 @@ class TestDisabledTelemetry: "deploy:created", "deploy:pushed", ] + + +class TestReleaseAttribution: + """Every span this emitter produces must carry the running release. + + A span without ``crewai_version`` cannot be attributed to a version, so a + version-filtered question returns nothing for it rather than something + visibly wrong. Covers all eight spans this module emits, not only the ones + that were missing it, so a regression on the others is caught too. + """ + + @pytest.mark.parametrize( + ("method", "args"), + [ + ("deploy_signup_error_span", ()), + ("start_deployment_span", ("dep-123",)), + ("create_crew_deployment_span", ()), + ("get_crew_logs_span", ("dep-123", "deployment")), + ("remove_crew_span", ("dep-123",)), + ("feature_usage_span", ("memory:query",)), + ("flow_creation_span", ("ResearchFlow",)), + ("template_installed_span", ("my-template",)), + ], + ) + def test_span_records_the_release( + self, + telemetry: tuple[Telemetry, MagicMock], + method: str, + args: tuple[object, ...], + ) -> None: + instance, span = telemetry + sentinel = "0.0.0-release-sentinel" + + # Patched at the source module, not at crewai_core.telemetry: these + # methods import get_crewai_version inside the call, so a patch on the + # importing module would never be seen. An arbitrary sentinel also means + # a hard-coded literal cannot satisfy the assertion. + with patch("crewai_core.version.get_crewai_version", return_value=sentinel): + getattr(instance, method)(*args) + + assert _attributes(span).get("crewai_version") == sentinel, ( + f"{method} did not record the value returned by get_crewai_version()" + ) diff --git a/lib/crewai/src/crewai/telemetry/telemetry.py b/lib/crewai/src/crewai/telemetry/telemetry.py index cd60505b2..3ed1d388d 100644 --- a/lib/crewai/src/crewai/telemetry/telemetry.py +++ b/lib/crewai/src/crewai/telemetry/telemetry.py @@ -508,6 +508,7 @@ class Telemetry: tracer = self.provider.get_tracer(TRACER_NAME) created_span = tracer.start_span("Task Created") + self._add_attribute(created_span, "crewai_version", version("crewai")) add_crew_and_task_attributes(created_span, crew, task, self._add_attribute) @@ -543,6 +544,7 @@ class Telemetry: close_span(created_span) span = tracer.start_span("Task Execution") + self._add_attribute(span, "crewai_version", version("crewai")) add_crew_and_task_attributes(span, crew, task, self._add_attribute) @@ -749,6 +751,7 @@ class Telemetry: def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Deploy Signup Error") + self._add_attribute(span, "crewai_version", version("crewai")) close_span(span) self._safe_telemetry_operation(_operation) @@ -763,6 +766,7 @@ class Telemetry: def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Start Deployment") + self._add_attribute(span, "crewai_version", version("crewai")) if uuid: self._add_attribute(span, "uuid", uuid) close_span(span) @@ -775,6 +779,7 @@ class Telemetry: def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Create Crew Deployment") + self._add_attribute(span, "crewai_version", version("crewai")) close_span(span) self._safe_telemetry_operation(_operation) @@ -792,6 +797,7 @@ class Telemetry: def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Get Crew Logs") + self._add_attribute(span, "crewai_version", version("crewai")) self._add_attribute(span, "log_type", log_type) if uuid: self._add_attribute(span, "uuid", uuid) @@ -809,6 +815,7 @@ class Telemetry: def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Remove Crew") + self._add_attribute(span, "crewai_version", version("crewai")) if uuid: self._add_attribute(span, "uuid", uuid) close_span(span) @@ -985,6 +992,7 @@ class Telemetry: def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Flow Plotting") + self._add_attribute(span, "crewai_version", version("crewai")) self._add_attribute(span, "flow_name", flow_name) self._add_attribute(span, "node_names", json.dumps(node_names)) close_span(span) @@ -1211,6 +1219,7 @@ class Telemetry: def _operation() -> None: tracer = self.provider.get_tracer(TRACER_NAME) span = tracer.start_span("Human Feedback") + self._add_attribute(span, "crewai_version", version("crewai")) self._add_attribute(span, "event_type", event_type) self._add_attribute(span, "has_routing", has_routing) self._add_attribute(span, "num_outcomes", num_outcomes) diff --git a/lib/crewai/tests/telemetry/test_telemetry.py b/lib/crewai/tests/telemetry/test_telemetry.py index 70b0fd0e8..f555a562f 100644 --- a/lib/crewai/tests/telemetry/test_telemetry.py +++ b/lib/crewai/tests/telemetry/test_telemetry.py @@ -1,10 +1,14 @@ +import ast +import inspect import os +from pathlib import Path import threading from unittest.mock import Mock, patch import pytest from crewai import Agent, Crew, Task from crewai.telemetry import Telemetry +from crewai_core.telemetry import Telemetry as CoreTelemetry from opentelemetry.sdk.trace import TracerProvider @@ -90,7 +94,7 @@ def test_flow_execution_span_records_crewai_version(): "crewai.telemetry.telemetry.TracerProvider", return_value=Mock(get_tracer=Mock(return_value=tracer)), ), - patch("crewai.telemetry.telemetry.version", return_value="9.9.9"), + patch("crewai.telemetry.telemetry.version", return_value=_EMITTED_VERSION), ): telemetry = Telemetry() telemetry.flow_execution_span("ResearchFlow", ["start", "finish"]) @@ -308,6 +312,11 @@ def test_event_listener_tracks_hook_dispatched_events(): ) +# The version _emit injects. Assertions compare against this exact value, so a +# hard-coded literal in an emitter cannot satisfy them. +_EMITTED_VERSION = "9.9.9" + + def _emit(method: str, *args, **kwargs): """Run one telemetry span method against a mocked tracer. @@ -382,3 +391,123 @@ def test_paused_and_method_failed_record_flow_and_origin() -> None: _tracer, span = _emit(method, "ResearchFlow", "internal") span.set_attribute.assert_any_call("flow_name", "ResearchFlow") span.set_attribute.assert_any_call("origin", "internal") + + +def _version_attr(span) -> str | None: + """The crewai_version value recorded on a mocked span, if any.""" + for call in span.set_attribute.call_args_list: + if call.args and call.args[0] == "crewai_version": + return call.args[1] + return None + + +@pytest.mark.parametrize( + ("method", "args"), + [ + ("flow_plotting_span", ("ResearchFlow", ["step_a", "step_b"])), + ("deploy_signup_error_span", ()), + ("start_deployment_span", ("dep-123",)), + ("create_crew_deployment_span", ()), + ("get_crew_logs_span", ("dep-123", "deployment")), + ("remove_crew_span", ("dep-123",)), + ("human_feedback_span", ("requested", False)), + ], +) +def test_span_records_the_crewai_version(method: str, args: tuple) -> None: + """Version-filtered queries silently drop any span kind missing this. + + Without it a release cannot be attributed for that span, so version-adoption + and per-release regression analysis are blind to it. + """ + _tracer, span = _emit(method, *args) + + # Exact equality with the value _emit injected: "looks like a version" would + # also accept a hard-coded literal in the emitter. + assert _version_attr(span) == _EMITTED_VERSION, ( + f"{method} did not record the value returned by version('crewai')" + ) + + +def test_task_spans_record_the_crewai_version() -> None: + """Task Created and Task Execution are the highest-volume span kinds. + + They are emitted together by task_started, and both were missing the + version - so every version-filtered task metric returned nothing. + """ + agent = Agent(role="R", goal="G", backstory="B") + task = Task(description="D", expected_output="E", agent=agent) + crew = Crew(agents=[agent], tasks=[task]) + + tracer, span = _emit("task_started", crew, task) + + emitted = [c.args[0] for c in tracer.start_span.call_args_list] + assert emitted == ["Task Created", "Task Execution"] + + # The harness hands the same mock back for both start_span calls, so the + # attribute writes accumulate: one crewai_version per span emitted. + versions = [ + c.args[1] + for c in span.set_attribute.call_args_list + if c.args and c.args[0] == "crewai_version" + ] + assert versions == [_EMITTED_VERSION, _EMITTED_VERSION], ( + f"expected one version per task span, got {versions}" + ) + + +def _calls(node: ast.AST, attr: str, key: str | None = None) -> int: + """Count calls to ``.attr(...)`` beneath a node, optionally keyed on arg 2. + + Used to compare how many spans a method opens against how many of them it + records ``crewai_version`` on. + """ + total = 0 + for sub in ast.walk(node): + if not isinstance(sub, ast.Call): + continue + func = sub.func + if not isinstance(func, ast.Attribute) or func.attr != attr: + continue + if key is None: + total += 1 + elif len(sub.args) >= 2: + named = sub.args[1] + if isinstance(named, ast.Constant) and named.value == key: + total += 1 + return total + + +def test_every_span_records_the_crewai_version() -> None: + """Regression guard for span kinds added later, in BOTH emitters. + + Enumerating the source rather than emitting all 32 spans: the point is to + fail when someone adds a new span without the version, which a fixed list of + behavioural cases cannot do. + + Counts rather than merely detects. A method that opens two spans and records + the version on only one of them must fail - ``task_started`` is exactly that + shape, so "the method mentions crewai_version somewhere" is not enough. + """ + shortfalls: list[str] = [] + for cls in (Telemetry, CoreTelemetry): + path = Path(inspect.getfile(cls)) + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + # The nested closure is reached via its enclosing method, whose name + # is the one a reader needs in the failure message. + if not isinstance(node, ast.FunctionDef) or node.name == "_operation": + continue + spans = _calls(node, "start_span") + if not spans: + continue + versions = _calls(node, "_add_attribute", "crewai_version") + if versions < spans: + shortfalls.append( + f"{path.name}::{node.name} " + f"({spans} span(s), {versions} version attribute(s))" + ) + + assert not shortfalls, ( + "these methods open more spans than they record crewai_version on: " + + ", ".join(sorted(shortfalls)) + )