fix: emit coding_agent as a span attribute, not a Resource attribute

The previous commit set coding_agent on the OTel Resource. Verified
against the telemetry ClickHouse instance that this would have silently
produced nothing: across 2,000,000 sampled spans the `process` column
contains exactly one key, `serviceName`, and no row has more than one.
The ingestion pipeline discards every other resource attribute.

Replace it with CommonAttributesSpanProcessor, whose on_start hook
applies the attribute to every span the provider emits. Span attributes
are preserved through ingestion and land in the `tags` array alongside
crew_key and crewai_version, which is where existing extraction reads
from.

- Add CommonAttributesSpanProcessor, a SpanProcessor that applies a fixed
  attribute set at span start. Attribute application is wrapped so a
  failure can never propagate into user execution.
- Remove the now-redundant explicit coding_agent attributes from the Crew
  Created and Flow Creation spans; the processor covers all spans.
- Replace the resource-attribute test with an end-to-end one that exports
  four differently-named spans through a real TracerProvider and asserts
  coding_agent survives on each, and that it is NOT on the resource.
- Add a test asserting the processor swallows attribute-application
  errors.

Verified at runtime that the real Telemetry provider installs the
processor and that an emitted span carries {'coding_agent':
'claude_code', 'crew_key': 'abc'} with resource {'service.name':
'crewAI-telemetry'}.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UNumDnNbiyw3pv1WakAe6t

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)

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):