fix(telemetry): record task and crew failures instead of reporting them as OK

Task and crew failures were indistinguishable from successes in telemetry,
which is why error_count is zero for every month in the downstream
aggregates rather than merely low.

Three separate defects:

1. Task failures were recorded as successes. TaskFailedEvent routed to
   Telemetry.task_ended, which calls close_span() - and close_span
   unconditionally sets StatusCode.OK. Every failed task was exported as
   OK, so no downstream query could ever count one.

2. Crew failures were not recorded at all, and leaked their span.
   on_crew_failed never touched telemetry, so a crew that raised left
   _execution_span open: never ended, never exported. The failure was
   invisible and the span was lost entirely.

3. Some task failures leaked their span too. on_task_failed only ended the
   span when source.agent.crew was present, so a task failing without one
   was popped from the span map and never closed.

Changes:
- Add close_span_with_error(), which sets StatusCode.ERROR and optionally
  records an error_type attribute.
- Add Telemetry.task_failed() and Telemetry.crew_failed(); crew_failed
  clears _execution_span so it cannot be double-closed.
- Wire TaskFailedEvent and CrewKickoffFailedEvent to them, closing spans
  unconditionally so neither can leak.
- Add optional error_type to TaskFailedEvent and CrewKickoffFailedEvent,
  populated with type(e).__name__ at the four emit sites. Defaults to None,
  so existing callers are unaffected.

PII: only the exception *class name* is recorded, never the message, which
routinely contains prompts, model output, and credentials. close_span_with_error
drops any value failing str.isidentifier(), so a message cannot be recorded
even if passed by mistake. Tests assert this against six message-shaped
inputs.

Tests: new tests/telemetry/test_failure_instrumentation.py (16 tests) covering
error status, the success/failure distinction, the PII guard, span-leak
regressions for both task and crew, and event backwards compatibility. The
module sets OTEL_SDK_DISABLED explicitly - the suite runs with the SDK
disabled and the root conftest pops the variable on teardown, so tests that
need real spans must not rely on that leak.

Note: total_duration_ms is a separate, pipeline-side issue. The raw `duration`
column is a Go-style string ("2.026641s"), so toInt64OrZero() yields 0 for
99.99% of rows. That fix belongs in the ClickHouse materialized views, not here.

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

lib/crewai/src/crewai/crew.py

@@ -1065,6 +1065,7 @@ class Crew(FlowTrackable, BaseModel):
                 self,
                 CrewKickoffFailedEvent(
                     error=str(e),
+                    error_type=type(e).__name__,
                     crew_name=self.name,
                     started_event_id=self._kickoff_event_id,
                 ),
@@ -1279,6 +1280,7 @@ class Crew(FlowTrackable, BaseModel):
                 self,
                 CrewKickoffFailedEvent(
                     error=str(e),
+                    error_type=type(e).__name__,
                     crew_name=self.name,
                     started_event_id=self._kickoff_event_id,
                 ),

lib/crewai/src/crewai/events/event_listener.py

@@ -198,6 +198,11 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(CrewKickoffFailedEvent)
         def on_crew_failed(source: Any, event: CrewKickoffFailedEvent) -> None:
+            # Previously this handler never touched telemetry, so a crew that
+            # raised left its execution span open: never ended, never exported,
+            # and the failure invisible downstream.
+            self._telemetry.crew_failed(source, event.error_type)
+
             self.formatter.handle_crew_status(
                 event.crew_name or "Crew",
                 source.id,
@@ -261,8 +266,10 @@ class EventListener(BaseEventListener):
         def on_task_failed(source: Any, event: TaskFailedEvent) -> None:
             span = self.execution_spans.pop(source, None)
             if span:
-                if source.agent and source.agent.crew:
+                # Closed unconditionally: previously the span was only ended
-                    self._telemetry.task_ended(span, source, source.agent.crew)
+                # when source.agent.crew was present, so any task that failed
+                # without one leaked its span and was never exported.
+                self._telemetry.task_failed(span, source, event.error_type)
 
             task_name = get_task_name(source)
             self.formatter.handle_task_status(

lib/crewai/src/crewai/events/types/crew_events.py

@@ -52,6 +52,13 @@ class CrewKickoffFailedEvent(CrewBaseEvent):
     """Event emitted when a crew fails to complete execution"""
 
     error: str
+    error_type: str | None = None
+    """Exception class name (e.g. "ValidationError").
+
+    Kept separate from ``error`` so telemetry can record what kind of failure
+    occurred without ever touching the message, which routinely contains
+    prompts, model output, or credentials.
+    """
     type: Literal["crew_kickoff_failed"] = "crew_kickoff_failed"
 
 

lib/crewai/src/crewai/events/types/task_events.py

@@ -49,6 +49,13 @@ class TaskFailedEvent(BaseEvent):
     """Event emitted when a task fails"""
 
     error: str
+    error_type: str | None = None
+    """Exception class name (e.g. "ValidationError").
+
+    Kept separate from ``error`` so telemetry can record what kind of failure
+    occurred without ever touching the message, which routinely contains
+    prompts, model output, or credentials.
+    """
     type: Literal["task_failed"] = "task_failed"
     task: Any | None = None
 

lib/crewai/src/crewai/task.py

@@ -797,7 +797,10 @@ class Task(BaseModel):
             return task_output
         except Exception as e:
             self.end_time = datetime.datetime.now()
-            crewai_event_bus.emit(self, TaskFailedEvent(error=str(e), task=self))
+            crewai_event_bus.emit(
+                self,
+                TaskFailedEvent(error=str(e), error_type=type(e).__name__, task=self),
+            )
             raise e
         finally:
             clear_task_files(self.id)
@@ -953,7 +956,10 @@ class Task(BaseModel):
             return task_output
         except Exception as e:
             self.end_time = datetime.datetime.now()
-            crewai_event_bus.emit(self, TaskFailedEvent(error=str(e), task=self))
+            crewai_event_bus.emit(
+                self,
+                TaskFailedEvent(error=str(e), error_type=type(e).__name__, task=self),
+            )
             raise e
         finally:
             clear_task_files(self.id)

lib/crewai/src/crewai/telemetry/telemetry.py

@@ -51,6 +51,7 @@ from crewai.telemetry.utils import (
     add_crew_and_task_attributes,
     add_crew_attributes,
     close_span,
+    close_span_with_error,
 )
 from crewai.utilities.i18n import I18N_DEFAULT
 from crewai.utilities.logger_utils import suppress_warnings
@@ -571,6 +572,30 @@ class Telemetry:
 
         self._safe_telemetry_operation(_operation)
 
+    def task_failed(
+        self, span: Span, task: Task, error_type: str | None = None
+    ) -> None:
+        """Records that a task execution failed and closes its span with ERROR.
+
+        Previously failures were routed through task_ended, which closes every
+        span as OK - making failed and successful tasks indistinguishable
+        downstream and leaving error counts permanently at zero.
+
+        Args:
+            span: The OpenTelemetry span tracking the task execution.
+            task: The task that failed.
+            error_type: Exception class name. The error message is never
+                recorded - it routinely contains prompts and model output.
+        """
+
+        def _operation() -> None:
+            if hasattr(task, "fingerprint") and task.fingerprint:
+                self._add_attribute(span, "task_fingerprint", task.fingerprint.uuid_str)
+
+            close_span_with_error(span, error_type)
+
+        self._safe_telemetry_operation(_operation)
+
     def tool_repeated_usage(self, llm: Any, tool_name: str, attempts: int) -> None:
         """Records when a tool is used repeatedly, which might indicate an issue.
 
@@ -922,6 +947,28 @@ class Telemetry:
         if crew.share_crew:
             self._safe_telemetry_operation(_operation)
 
+    def crew_failed(self, crew: Any, error_type: str | None = None) -> None:
+        """Records that a crew execution failed and closes its span.
+
+        Without this, a crew that raises leaves its execution span open: it is
+        never ended, never exported, and the failure is invisible downstream.
+
+        Args:
+            crew: The crew whose execution failed.
+            error_type: Exception class name. The error message is never
+                recorded - it routinely contains prompts and model output.
+        """
+
+        def _operation() -> None:
+            span = getattr(crew, "_execution_span", None)
+            if span is None:
+                return
+            self._add_attribute(span, "crewai_version", version("crewai"))
+            close_span_with_error(span, error_type)
+            crew._execution_span = None
+
+        self._safe_telemetry_operation(_operation)
+
     def _add_attribute(self, span: Span, key: str, value: Any) -> None:
         """Add an attribute to a span.
 

lib/crewai/src/crewai/telemetry/utils.py

@@ -111,3 +111,23 @@ def close_span(span: Span) -> None:
     """
     span.set_status(Status(StatusCode.OK))
     span.end()
+
+
+def close_span_with_error(span: Span, error_type: str | None = None) -> None:
+    """Set span status to ERROR and end it.
+
+    Used for spans representing work that failed, so failures are
+    distinguishable from successes downstream. Only the exception's *type* is
+    recorded - never the message, which routinely contains prompts, model
+    output, or credentials.
+
+    Args:
+        span: The span to close.
+        error_type: Exception class name (e.g. "ValidationError"). Anything
+            that is not a plain identifier is discarded rather than recorded,
+            so a message can never be passed in by mistake.
+    """
+    span.set_status(Status(StatusCode.ERROR))
+    if error_type and error_type.isidentifier():
+        span.set_attribute("error_type", error_type)
+    span.end()

lib/crewai/tests/telemetry/test_failure_instrumentation.py

@@ -0,0 +1,210 @@
+"""Tests that failed executions are recorded as failures, not successes.
+
+Regression coverage for telemetry that reported every task as OK, leaving
+downstream error counts permanently at zero.
+"""
+
+from unittest.mock import Mock
+
+import pytest
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
+    InMemorySpanExporter,
+)
+from opentelemetry.trace import StatusCode
+
+from crewai.telemetry.utils import close_span, close_span_with_error
+
+
+@pytest.fixture(autouse=True)
+def enable_otel_sdk(monkeypatch):
+    """Ensure the OTel SDK is active for these tests.
+
+    The suite runs with OTEL_SDK_DISABLED=true, which makes TracerProvider hand
+    out non-recording spans that are never exported. Set explicitly rather than
+    relying on the root conftest teardown, which pops the variable and would
+    otherwise leave only the first test in a session running against a
+    disabled SDK.
+    """
+    monkeypatch.delenv("OTEL_SDK_DISABLED", raising=False)
+    monkeypatch.delenv("CREWAI_DISABLE_TELEMETRY", raising=False)
+    monkeypatch.delenv("CREWAI_DISABLE_TRACKING", raising=False)
+
+
+@pytest.fixture
+def exporter():
+    exp = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exp))
+    # yield rather than return: the generator frame keeps `provider` alive for
+    # the test. If it is collected, its processor shuts down and spans are lost.
+    yield exp, provider.get_tracer("test")
+
+
+def test_close_span_with_error_sets_error_status(exporter):
+    exp, tracer = exporter
+
+    close_span_with_error(tracer.start_span("Task Execution"), "ValidationError")
+
+    span = exp.get_finished_spans()[0]
+    assert span.status.status_code is StatusCode.ERROR
+    assert span.attributes["error_type"] == "ValidationError"
+
+
+def test_successful_and_failed_spans_are_distinguishable(exporter):
+    """The whole point: a downstream count of failures must be possible."""
+    exp, tracer = exporter
+
+    close_span(tracer.start_span("Task Execution"))
+    close_span_with_error(tracer.start_span("Task Execution"), "TimeoutError")
+    close_span(tracer.start_span("Task Execution"))
+
+    spans = exp.get_finished_spans()
+    failed = [s for s in spans if s.status.status_code is StatusCode.ERROR]
+    assert len(spans) == 3
+    assert len(failed) == 1
+    assert failed[0].attributes["error_type"] == "TimeoutError"
+
+
+@pytest.mark.parametrize(
+    "not_an_identifier",
+    [
+        "Rate limit exceeded for gpt-4o",
+        "API key sk-live-1234 is invalid",
+        "connection to db://user:pass@host failed",
+        "",
+        "  ",
+        "429",
+    ],
+)
+def test_error_message_can_never_be_recorded(exporter, not_an_identifier):
+    """PII guard: only identifier-shaped values survive.
+
+    Error messages routinely contain prompts, model output, and credentials.
+    Passing one where an exception class name belongs must record nothing.
+    """
+    exp, tracer = exporter
+
+    close_span_with_error(tracer.start_span("Task Execution"), not_an_identifier)
+
+    span = exp.get_finished_spans()[0]
+    assert span.status.status_code is StatusCode.ERROR
+    assert "error_type" not in (span.attributes or {})
+
+
+def test_error_type_is_optional(exporter):
+    exp, tracer = exporter
+
+    close_span_with_error(tracer.start_span("Task Execution"))
+
+    span = exp.get_finished_spans()[0]
+    assert span.status.status_code is StatusCode.ERROR
+    assert "error_type" not in (span.attributes or {})
+
+
+def test_real_exception_class_names_are_accepted(exporter):
+    """Every builtin exception name is a valid identifier, so none are dropped."""
+    exp, tracer = exporter
+
+    for exc in (ValueError, TimeoutError, KeyError, RuntimeError, ConnectionError):
+        close_span_with_error(tracer.start_span("Task Execution"), exc.__name__)
+
+    recorded = [s.attributes["error_type"] for s in exp.get_finished_spans()]
+    assert recorded == [
+        "ValueError",
+        "TimeoutError",
+        "KeyError",
+        "RuntimeError",
+        "ConnectionError",
+    ]
+
+
+def test_task_failed_closes_span_with_error():
+    from crewai.telemetry.telemetry import Telemetry
+
+    exp = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exp))
+
+    telemetry = Telemetry()
+    telemetry.ready = True
+    span = provider.get_tracer("test").start_span("Task Execution")
+
+    telemetry.task_failed(span, Mock(fingerprint=None), "ValueError")
+
+    finished = exp.get_finished_spans()[0]
+    assert finished.status.status_code is StatusCode.ERROR
+    assert finished.attributes["error_type"] == "ValueError"
+
+
+def test_crew_failed_closes_leaked_execution_span():
+    """A crew that raises must not leave its span open and unexported."""
+    from crewai.telemetry.telemetry import Telemetry
+
+    exp = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exp))
+
+    telemetry = Telemetry()
+    telemetry.ready = True
+
+    crew = Mock()
+    crew._execution_span = provider.get_tracer("test").start_span("Crew Execution")
+
+    telemetry.crew_failed(crew, "RuntimeError")
+
+    finished = exp.get_finished_spans()
+    assert len(finished) == 1, "span was never ended - it would never be exported"
+    assert finished[0].status.status_code is StatusCode.ERROR
+    assert finished[0].attributes["error_type"] == "RuntimeError"
+    assert crew._execution_span is None
+
+
+def test_crew_failed_is_safe_when_no_span_exists():
+    """share_crew=False crews have no execution span; this must not raise."""
+    from crewai.telemetry.telemetry import Telemetry
+
+    telemetry = Telemetry()
+    telemetry.ready = True
+
+    crew = Mock()
+    crew._execution_span = None
+
+    telemetry.crew_failed(crew, "RuntimeError")
+
+
+def test_task_failed_event_carries_error_type():
+    """The exception class must reach the event without the message."""
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    try:
+        raise TimeoutError("request to gpt-4o timed out after 60s")
+    except TimeoutError as e:
+        event = TaskFailedEvent(error=str(e), error_type=type(e).__name__, task=None)
+
+    assert event.error_type == "TimeoutError"
+    assert "gpt-4o" not in event.error_type
+
+
+def test_crew_kickoff_failed_event_carries_error_type():
+    from crewai.events.types.crew_events import CrewKickoffFailedEvent
+
+    try:
+        raise ValueError("bad input: {'api_key': 'sk-live-1234'}")
+    except ValueError as e:
+        event = CrewKickoffFailedEvent(
+            error=str(e), error_type=type(e).__name__, crew_name="TestCrew"
+        )
+
+    assert event.error_type == "ValueError"
+    assert "sk-live" not in event.error_type
+
+
+def test_error_type_defaults_to_none_for_backwards_compatibility():
+    """Existing callers that omit error_type must keep working."""
+    from crewai.events.types.crew_events import CrewKickoffFailedEvent
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    assert TaskFailedEvent(error="boom", task=None).error_type is None
+    assert CrewKickoffFailedEvent(error="boom", crew_name="C").error_type is None