diff --git a/lib/crewai/src/crewai/telemetry/telemetry.py b/lib/crewai/src/crewai/telemetry/telemetry.py index bdb6b60cc..503e2f106 100644 --- a/lib/crewai/src/crewai/telemetry/telemetry.py +++ b/lib/crewai/src/crewai/telemetry/telemetry.py @@ -21,11 +21,12 @@ import threading from typing import TYPE_CHECKING, Any from opentelemetry import trace +from opentelemetry.context import Context from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( OTLPSpanExporter, ) from opentelemetry.sdk.resources import SERVICE_NAME, Resource -from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace import SpanProcessor, TracerProvider from opentelemetry.sdk.trace.export import ( BatchSpanProcessor, SpanExportResult, @@ -88,6 +89,57 @@ class SafeOTLPSpanExporter(OTLPSpanExporter): return SpanExportResult.FAILURE +class CommonAttributesSpanProcessor(SpanProcessor): + """Applies a fixed set of attributes to every span at start. + + Used for process-wide context that should appear on all spans (e.g. which + AI coding assistant is running the process) without each span-emitting + method having to set it. Attributes are applied as span attributes rather + than Resource attributes because the ingestion pipeline preserves only + serviceName from the resource. + """ + + def __init__(self, attributes: dict[str, str]) -> None: + """Initialize the processor. + + Args: + attributes: Attributes applied to every span. Values must not + contain user data - this is process-wide context only. + """ + self._attributes = attributes + + def on_start( + self, span: Span, parent_context: Context | None = None + ) -> None: + """Apply the common attributes to a span as it starts. + + Args: + span: The span being started. + parent_context: Parent context, unused. + """ + try: + span.set_attributes(self._attributes) + except Exception: # noqa: S110 - telemetry must never break execution + pass + + def on_end(self, span: Any) -> None: + """No-op; export is handled by the batch processor.""" + + def shutdown(self) -> None: + """No-op; this processor holds no resources.""" + + def force_flush(self, timeout_millis: int = 30000) -> bool: + """No-op flush. + + Args: + timeout_millis: Unused. + + Returns: + Always True. + """ + return True + + class Telemetry: """Handle anonymous telemetry for the CrewAI package. @@ -123,19 +175,20 @@ class Telemetry: return try: - # coding_agent is set on the Resource so it is attached to *every* - # span this provider emits, without per-method duplication. The value - # is one of a fixed set of literals from detect_coding_agent() and - # never contains environment values or any user data. self.resource = Resource( - attributes={ - SERVICE_NAME: CREWAI_TELEMETRY_SERVICE_NAME, - "coding_agent": detect_coding_agent(), - }, + attributes={SERVICE_NAME: CREWAI_TELEMETRY_SERVICE_NAME}, ) with suppress_warnings(): self.provider = TracerProvider(resource=self.resource) + # coding_agent is applied as a *span attribute* via on_start, not as + # a Resource attribute: the ingestion pipeline only preserves + # serviceName from the resource, so anything else set there is + # dropped before it reaches storage. Span attributes are preserved. + self.provider.add_span_processor( + CommonAttributesSpanProcessor({"coding_agent": detect_coding_agent()}) + ) + processor = BatchSpanProcessor( SafeOTLPSpanExporter( endpoint=f"{CREWAI_TELEMETRY_BASE_URL}/v1/traces", @@ -293,7 +346,6 @@ class Telemetry: version("crewai"), ) self._add_attribute(span, "python_version", platform.python_version()) - self._add_attribute(span, "coding_agent", detect_coding_agent()) add_crew_attributes(span, crew, self._add_attribute) self._add_attribute(span, "crew_process", crew.process) self._add_attribute(span, "crew_memory", crew.memory) @@ -963,7 +1015,6 @@ class Telemetry: span = tracer.start_span("Flow Creation") self._add_attribute(span, "crewai_version", version("crewai")) self._add_attribute(span, "flow_name", flow_name) - self._add_attribute(span, "coding_agent", detect_coding_agent()) close_span(span) self._safe_telemetry_operation(_operation) diff --git a/lib/crewai/tests/telemetry/test_coding_agent_detection.py b/lib/crewai/tests/telemetry/test_coding_agent_detection.py index cfc047555..3f7761ed8 100644 --- a/lib/crewai/tests/telemetry/test_coding_agent_detection.py +++ b/lib/crewai/tests/telemetry/test_coding_agent_detection.py @@ -127,31 +127,54 @@ def test_known_agents_contains_no_pii_shaped_values(): assert len(name) <= 32, name -def test_coding_agent_attached_to_telemetry_resource(clean_env, monkeypatch): - """The attribute must land on the Resource, so it reaches every span.""" - import os - from unittest.mock import patch +def test_coding_agent_lands_on_every_exported_span(clean_env): + """End-to-end: the attribute must appear as a *span attribute* on any span. - from crewai.telemetry.telemetry import Telemetry + It cannot be a Resource attribute - the ingestion pipeline preserves only + serviceName from the resource, so anything else set there is dropped before + it reaches storage. This test exports through a real TracerProvider and + asserts the attribute survives on arbitrary spans. + """ + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) - clean_env.setenv("CLAUDECODE", "1") + from crewai.telemetry.telemetry import CommonAttributesSpanProcessor - with ( - patch.dict( - os.environ, - { - "CREWAI_DISABLE_TELEMETRY": "false", - "CREWAI_DISABLE_TRACKING": "false", - "OTEL_SDK_DISABLED": "false", - }, - ), - patch("crewai.telemetry.telemetry.TracerProvider"), - ): - telemetry = Telemetry() - telemetry._initialized = False - telemetry.__init__() + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor( + CommonAttributesSpanProcessor({"coding_agent": "claude_code"}) + ) + provider.add_span_processor(SimpleSpanProcessor(exporter)) - assert telemetry.resource.attributes["coding_agent"] == "claude_code" + tracer = provider.get_tracer("crewai.telemetry") + for name in ("Crew Created", "Task Execution", "Tool Usage", "Feature Usage"): + span = tracer.start_span(name) + span.end() + + exported = exporter.get_finished_spans() + assert len(exported) == 4 + for span in exported: + assert span.attributes["coding_agent"] == "claude_code", span.name + + # It must be a span attribute, not a resource attribute, or ingestion drops it. + assert "coding_agent" not in exported[0].resource.attributes + + +def test_common_attributes_processor_never_breaks_span_creation(clean_env): + """A failure applying attributes must not propagate into user execution.""" + from crewai.telemetry.telemetry import CommonAttributesSpanProcessor + + class ExplodingSpan: + def set_attributes(self, _): + raise RuntimeError("boom") + + CommonAttributesSpanProcessor({"coding_agent": "cursor"}).on_start( + ExplodingSpan() # type: ignore[arg-type] + ) def test_coding_agent_span_emits_once(clean_env, monkeypatch):