fix(events): record task failures as failures, not as successes (#7073)

* fix(telemetry): record task failures as failures, not as successes

close_span() sets StatusCode.OK unconditionally, and TaskFailedEvent was routed
to Telemetry.task_ended, which calls it. Every failed task was therefore
exported as OK, which is why error_count downstream is not merely low but
exactly zero: 240.0M task executions across 13 months in
crew_task_executions_daily_target, error_count = 0 in every one of them.

The same line had a second defect. The span was only ended when
source.agent.crew was present, so a task failing without one was popped from the
span map and then never closed - never ended, never exported, invisible rather
than mislabelled. task_failed takes no crew (it reads nothing off one), so that
condition disappears rather than being widened.

Only the exception class name is recorded, never the message, which routinely
contains prompts, model output, file paths and credentials.
close_span_with_error drops any value failing str.isidentifier(), so a message
cannot be recorded even if one is passed by mistake.

This is the task half of closed PR #6781, re-cut onto main as that PR asked for.
The crew half is deliberately left out: crew_execution_span() returns None unless
share_crew=True, so crew._execution_span is None for nearly every user and a
crew-failure handler would exit immediately for the default population.

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

* fix(telemetry): take the exception class for error_type, not a free-form string

cursor and CodeRabbit both flagged the sanitization, and they were right: the
package already had a stronger convention and this change had not used it.
Telemetry._safe_error_type takes the exception *class*, and its docstring says
why in as many words - "a single-word message such as 'secret_token' is itself a
valid identifier", so filtering a string with isidentifier() is not enough. The
ported code predates that helper and reinvented the weaker check.

TaskFailedEvent.error_type is now type[BaseException] | None, so pydantic itself
rejects a message before any of our code runs, and task_failed routes it through
_safe_error_type. The identifier check in close_span_with_error stays as the
second gate on the derived name, which is the role _safe_error_type's docstring
already describes.

Also adds producer-level tests, which CodeRabbit correctly identified as missing:
every earlier test constructed TaskFailedEvent directly, so a regression in the
two emit sites this change touches in task.py would have passed the whole suite.
The sync and async producers are driven through Task._execute_core and
Task._aexecute_core with a distinctive exception class, and each patches a
different agent method (execute_task vs aexecute_task), which is why they can
regress independently. Verified by dropping error_type from both producers: all
three new tests fail, and pass again when restored.

Removes an unused `import os` left behind when the fixture was rewritten.

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

* test(telemetry): capture producer failures at the emit boundary, not via the bus

The producer tests subscribed a handler to crewai_event_bus and asserted on what
it received. That passed this file in isolation and every randomized local run,
then failed in CI inside a 621-test shard with zero events captured:

  FAILED tests/telemetry/test_task_failure_instrumentation.py::
    test_sync_producer_puts_the_exception_class_on_the_event
  assert 0 == 1  +  where 0 = len([])

task_failed is an "ending" event, and with an empty scope stack - there is no
real kickoff in these tests - dispatch is conditional on event-context state that
other tests in the same worker process can leave behind. Subscribing made the
assertion depend on the bus choosing to dispatch, which is not what these tests
are about: they are about what the producer in task.py constructs.

Patching crewai_event_bus.emit records the event unconditionally at the point the
producer hands it over, with no dispatch involved. Both producers ignore emit's
return value, so returning None is faithful.

Containment re-verified after the change: dropping error_type from both producers
fails exactly these three tests and nothing else.

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

* fix(events): keep TaskFailedEvent JSON-serializable with a class-valued error_type

error_type holds an exception class, which is not a JSON type, so
model_dump(mode="json") raised PydanticSerializationError for the whole event -
not just that field. Two real consumers depend on it: the checkpoint listener
dumps every event through EventRecord, and the tracing listener JSON-POSTs events
to AMP. A single task failure therefore took out checkpointing.

field_serializer with when_used="json" returns the class name. The "json" scope is
load-bearing: event_listener hands the live class to Telemetry.task_failed, which
needs it for _safe_error_type, so python-mode dumps must keep the class.

The annotation is a module-level _ExceptionClass alias rather than an inline
type[BaseException], because TaskFailedEvent declares a field named `type` which
shadows the builtin for the rest of the class body - inline, it raises TypeError
at import ("task_failed"[BaseException]) and mypy rejects it as "Variable ... is
not valid as a type". Quoting satisfies neither tool: ruff flags UP037 and mypy
still resolves it in the class scope.

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

* fix(events): let a dumped error_type restore, instead of degrading the event

The serializer added in the previous commit stopped model_dump(mode="json") from
raising, but nothing accepted the class-name string back. _resolve_event
(state/event_record.py:32-35) wraps cls.model_validate in a bare except and falls
back to BaseEvent, so restoring a checkpoint after a task failure silently
dropped the whole event -- including `error`, a plain string that would otherwise
have survived. Traded a loud failure for a quiet one.

Measured before: dumped error_type='ValueError' and error='boom', restored as
BaseEvent with neither attribute. After: restores as TaskFailedEvent with
error='boom' and error_type is ValueError.

A BeforeValidator resolves a name against real exception classes only -- builtins
first, then a walk of BaseException.__subclasses__(). So this does not reopen the
hole the class-typed field closes: "secret_token" resolves to nothing, is returned
unchanged, and is rejected by the field's own type. Asserted for secret_token,
sk_live_1234, dict and os.

A name whose class is not imported in this process still degrades, which is
deliberate: synthesising a class from an arbitrary string is the injection risk
this field exists to avoid.

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

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Vidit Ostwal <110953813+Vidit-Ostwal@users.noreply.github.com>

docs/edge/ar/telemetry.mdx

@@ -57,7 +57,7 @@ os.environ['OTEL_SDK_DISABLED'] = 'true'
 | نعم       | بيانات وصفية للمهمة | تشمل: مفتاح ومعرّف مُولّد عشوائياً، إعدادات تنفيذ منطقية (async_execution، human_input)، دور ومفتاح الوكيل المرتبط، قائمة أسماء الأدوات. كلها غير شخصية. |
 | نعم       | إحصائيات استخدام الأدوات | تشمل: اسم الأداة (يجب ألا يتضمن معلومات شخصية)، عدد محاولات الاستخدام (عدد صحيح)، سمات LLM المستخدمة. لا بيانات شخصية. |
 | نعم       | بيانات تنفيذ الاختبار | تشمل: مفتاح ومعرّف الطاقم المُولّد عشوائياً، عدد التكرارات، اسم النموذج المستخدم، درجة الجودة (عدد عشري)، وقت التنفيذ (بالثواني). كلها غير شخصية. |
-| نعم       | بيانات دورة حياة المهمة | تشمل: أوقات الإنشاء وبدء/انتهاء التنفيذ، معرّفات الطاقم والمهمة. مخزنة كنطاقات مع طوابع زمنية. لا بيانات شخصية. |
+| نعم       | بيانات دورة حياة المهمة | تشمل: أوقات الإنشاء وبدء/انتهاء التنفيذ، معرّفات الطاقم والمهمة، وما إذا نجحت المهمة أو فشلت. وعند فشل المهمة، يُسجَّل **اسم صنف** الاستثناء (مثل `TimeoutError`) بحيث يمكن عدّ حالات الفشل وتشخيصها — وليس رسالة الخطأ أبدًا، فهي قد تحتوي على مطالبات أو مخرجات نموذج أو مسارات ملفات أو بيانات اعتماد. مخزنة كنطاقات مع طوابع زمنية. لا بيانات شخصية. |
 | نعم       | سمات LLM | تشمل: الاسم، model_name، model، top_k، temperature، واسم فئة LLM. كلها بيانات تقنية غير شخصية. |
 | نعم       | إنشاء مشروع باستخدام CLI الخاص بـ CrewAI | تشمل: أن مشروعًا جديدًا أُنشئ عبر `crewai create`، ونوعه (`crew` أو `json_crew` أو `flow`)، ومعرّف المشروع الذي تم توليده لهذا المشروع الجديد وكُتب في ملف `pyproject.toml` الخاص به. وهو معرّف المشروع الجديد نفسه، ويُسجَّل بشكل منفصل عن `project_id` الخاص بالمجلد الذي شُغّل منه الأمر — وقد يختلفان. لا اسم مشروع، ولا محتويات ملفات، ولا شيفرة. لا بيانات شخصية. |
 | نعم       | محاولة نشر الطاقم باستخدام CLI الخاص بـ CrewAI | تشمل: حقيقة إجراء النشر ومعرّف الطاقم، وما إذا كان يحاول سحب السجلات، وما إذا بدأ النشر من أمر CLI أو من واجهة التشغيل TUI. لا تُسجَّل محتويات المشروع أو الطاقم. لا توجد بيانات شخصية. |

docs/edge/en/telemetry.mdx

@@ -57,7 +57,7 @@ own tracer provider, which is independent of the one described here.
 | Yes       | Task Metadata | Includes: randomly generated key and ID, boolean execution settings (async_execution, human_input), associated agent's role and key, list of tool names. All non-personal. |
 | Yes       | Tool Usage Statistics | Includes: tool name (should not include personal info), number of usage attempts (integer), LLM attributes used. No personal data. |
 | Yes       | Test Execution Data | Includes: crew's randomly generated key and ID, number of iterations, model name used, quality score (float), execution time (in seconds). All non-personal. |
-| Yes       | Task Lifecycle Data | Includes: creation and execution start/end times, crew and task identifiers. Stored as spans with timestamps. No personal data. |
+| Yes       | Task Lifecycle Data | Includes: creation and execution start/end times, crew and task identifiers, and whether the task succeeded or failed. When a task fails, the **class name** of the exception is recorded (for example `TimeoutError`) so failures can be counted and diagnosed — never the error message, which can contain prompts, model output, file paths or credentials. Stored as spans with timestamps. No personal data. |
 | Yes       | LLM Attributes | Includes: name, model_name, model, top_k, temperature, and class name of the LLM. All technical, non-personal data. |
 | Yes       | Project Creation using crewAI CLI | Includes: that a new project was scaffolded by `crewai create`, which kind it was (`crew`, `json_crew` or `flow`), and the project ID minted for that new project and written into its own `pyproject.toml`. That is the new project's own ID, recorded separately from the `project_id` of the directory the command was run from — the two can differ. No project name, no file contents, no code. No personal data. |
 | Yes       | Crew Deployment attempt using crewAI CLI | Includes: The fact a deploy is being made and crew id, whether it's trying to pull logs, and whether the deploy was started from a CLI command or from the run TUI. No project or crew contents. No personal data. |

docs/edge/ko/telemetry.mdx

@@ -55,7 +55,7 @@ provider로 등록하지 않습니다. 이를 통해 양방향이 분리됩니
 | 예      | 작업 메타데이터                            | 랜덤으로 생성된 키 및 ID, boolean 실행 설정(async_execution, human_input), 관련 에이전트 역할 및 키, 도구 이름 목록이 포함됩니다. 모두 비개인 정보입니다. |
 | 예      | 도구 사용 통계                             | 도구 이름(개인 정보 포함 불가), 사용 시도 횟수(정수), 사용된 LLM 속성이 포함됩니다. 개인 정보 없음. |
 | 예      | 테스트 실행 데이터                         | crew의 랜덤 생성 키와 ID, 반복 횟수, 사용된 모델명, 품질 점수(실수), 실행 시간(초 단위)이 포함됩니다. 모두 비개인 정보입니다. |
-| 예      | 작업 라이프사이클 데이터                   | 생성 및 실행 시작/종료 시각, crew 및 작업 식별자가 포함됩니다. 타임스탬프를 포함한 span으로 저장됩니다. 개인 정보 없음. |
+| 예      | 작업 라이프사이클 데이터                   | 생성 및 실행 시작/종료 시각, crew 및 작업 식별자, 그리고 작업의 성공 또는 실패 여부가 포함됩니다. 작업이 실패하면 실패를 집계하고 진단할 수 있도록 예외의 **클래스 이름**(예: `TimeoutError`)이 기록되며, 프롬프트·모델 출력·파일 경로·자격 증명이 포함될 수 있는 오류 메시지는 결코 기록되지 않습니다. 타임스탬프를 포함한 span으로 저장됩니다. 개인 정보 없음. |
 | 예      | LLM 속성                                  | LLM의 이름, model_name, 모델, top_k, temperature 및 클래스명이 포함됩니다. 모두 기술적이고 비개인 정보입니다. |
 | 예      | crewAI CLI를 통한 프로젝트 생성 | 포함 항목: `crewai create`로 새 프로젝트가 생성되었다는 사실, 그 종류(`crew`, `json_crew` 또는 `flow`), 그리고 그 새 프로젝트에 발급되어 해당 프로젝트의 `pyproject.toml`에 기록된 프로젝트 ID. 이는 새 프로젝트 자체의 ID이며, 명령을 실행한 디렉터리의 `project_id`와는 별개로 기록됩니다 — 두 값은 다를 수 있습니다. 프로젝트 이름, 파일 내용, 코드는 기록되지 않습니다. 개인 정보 없음. |
 | 예      | crewAI CLI를 통한 Crew 배포 시도 | 포함 항목: 배포가 시도되고 있다는 사실과 crew id, 로그를 가져오려고 하는지 여부, 그리고 배포가 CLI 명령에서 시작되었는지 실행 TUI에서 시작되었는지 여부. 프로젝트나 crew의 내용은 기록되지 않습니다. 개인 정보 없음. |

docs/edge/pt-BR/telemetry.mdx

@@ -57,7 +57,7 @@ por meio do próprio tracer provider, que é independente do descrito aqui.
 | Sim    | Metadados da Tarefa                       | Inclui: chave e ID gerados aleatoriamente, configurações de execução booleanas (async_execution, human_input), função e chave do agente associado, lista de nomes de ferramentas. Tudo não pessoal.      |
 | Sim    | Estatísticas de Uso de Ferramentas         | Inclui: nome da ferramenta (não deve incluir info pessoal), número de tentativas de uso (inteiro), atributos LLM utilizados. Sem dados pessoais.                        |
 | Sim    | Dados de Execução de Testes                | Inclui: chave e ID aleatórias do crew, número de iterações, nome do modelo usado, score de qualidade (float), tempo de execução (em segundos). Tudo não pessoal.         |
-| Sim    | Dados do Ciclo de Vida da Tarefa           | Inclui: horários de criação, início/fim de execução, identificadores de crew e tarefa. Armazenado como spans com timestamps. Sem dados pessoais.            |
+| Sim    | Dados do Ciclo de Vida da Tarefa           | Inclui: horários de criação, início/fim de execução, identificadores de crew e tarefa, e se a tarefa foi bem-sucedida ou falhou. Quando uma tarefa falha, o **nome da classe** da exceção é registrado (por exemplo `TimeoutError`) para que as falhas possam ser contadas e diagnosticadas — nunca a mensagem de erro, que pode conter prompts, saída do modelo, caminhos de arquivos ou credenciais. Armazenado como spans com timestamps. Sem dados pessoais.            |
 | Sim    | Atributos do LLM                           | Inclui: nome, model_name, model, top_k, temperatura e nome da classe do LLM. Todos técnicos, sem dados pessoais.                  |
 | Sim    | Criação de Projeto pelo CLI do crewAI | Inclui: o fato de um novo projeto ter sido criado por `crewai create`, de qual tipo ele é (`crew`, `json_crew` ou `flow`) e o ID de projeto gerado para esse novo projeto e gravado no `pyproject.toml` dele. É o ID do próprio projeto novo, registrado separadamente do `project_id` do diretório de onde o comando foi executado — os dois podem diferir. Sem nome de projeto, sem conteúdo de arquivos, sem código. Sem dados pessoais. |
 | Sim    | Tentativa de Deploy do Crew pelo CLI do crewAI | Inclui: O fato de um deploy estar sendo realizado e o crew id, se está tentando buscar logs, e se o deploy foi iniciado por um comando do CLI ou pela TUI de execução. Não inclui conteúdo do projeto ou do crew nem dados pessoais. |

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

@@ -266,8 +266,12 @@ 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:
-                    self._telemetry.task_ended(span, source, source.agent.crew)
+                # Routed to task_failed, not task_ended: the latter closes the span
+                # as OK, which is why a failed task was indistinguishable from a
+                # successful one. No longer conditional on source.agent.crew either -
+                # task_failed does not need a crew, and requiring one meant the span
+                # was popped and then never closed, so it was never exported at all.
+                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/task_events.py

@@ -1,9 +1,98 @@
-from typing import Any, Literal
+import sys
+from typing import Annotated, Any, Literal
+
+from pydantic import BeforeValidator, field_serializer
 
 from crewai.events.base_events import BaseEvent
 from crewai.tasks.task_output import TaskOutput
 
 
+def _resolve_exception_class(value: Any) -> Any:
+    """Turn a serialized ``module:qualname`` back into the class it names.
+
+    Needed because the JSON dump cannot write a class. Without this, restoring a
+    checkpoint fails ``TaskFailedEvent`` validation and ``_resolve_event`` silently
+    degrades the whole event to a bare ``BaseEvent`` -- losing ``error`` as well, which
+    is a plain string that would otherwise have survived.
+
+    **Resolution is exact, and by module rather than by name.** An earlier revision
+    matched on the bare ``__name__`` by walking ``BaseException.__subclasses__()``, which
+    could bind to a *different* class of the same name -- two modules, two scopes, or two
+    dynamically created classes can all share one. Reproduced by review with two distinct
+    ``type("DupErr", (Exception,), {})`` classes: the event restored as the wrong one.
+
+    **Nothing is imported here, and the lookup is by dict rather than by attribute.**
+    The module has to be in ``sys.modules`` already. Importing a module named in
+    serialized data would execute its top-level code -- a worse defect than the collision
+    it would fix, in a field that exists to keep untrusted strings out of telemetry.
+
+    ``vars()`` rather than ``getattr()`` is what makes that true, and the difference is
+    not theoretical: a module with a PEP 562 ``__getattr__`` runs it on any attribute
+    miss, and ``crewai.events`` itself is such a module -- its hook calls
+    ``importlib.import_module``. So ``getattr`` here would import a submodule named in
+    the serialized string, and any non-``AttributeError`` the hook raised would escape
+    validation and degrade the whole event, dropping ``error`` with it. A ``__dict__``
+    lookup consults no hook and runs no user code.
+
+    A serialized identity that does not resolve becomes ``None`` -- an unloaded module, a
+    ``<locals>`` qualname unreachable by attribute lookup, or a target that is not a
+    ``BaseException`` subclass. ``None`` rather than the original string, so the field
+    still validates and ``error`` survives the restore; the previous behaviour dropped
+    the entire event.
+
+    **A string with no ``":"`` is handled the other way, on purpose.** It cannot be
+    something this serializer wrote, so it is a caller passing a string where a class
+    belongs -- most likely ``str(error)``. That is returned unchanged and rejected by the
+    field's own type, which keeps the loud failure a silent ``None`` would hide. This is
+    the check that stops a message being recorded, and it is why the field is typed as a
+    class: a one-word message such as ``"secret_token"`` is itself a valid identifier, so
+    an ``isidentifier()`` gate on a plain string would let it through.
+    """
+    if not isinstance(value, str):
+        return value
+
+    module_name, separator, qualname = value.partition(":")
+    if not separator:
+        return value
+    if not qualname:
+        return None
+
+    resolved: Any = sys.modules.get(module_name)
+    if resolved is None:
+        return None
+    for attribute in qualname.split("."):
+        try:
+            namespace = vars(resolved)
+        except TypeError:
+            # No __dict__ to read: the qualname points at something that cannot
+            # hold a nested class, so there is nothing to resolve.
+            return None
+        resolved = namespace.get(attribute)
+        if resolved is None:
+            return None
+
+    if isinstance(resolved, type) and issubclass(resolved, BaseException):
+        return resolved
+    return None
+
+
+_ExceptionClass = Annotated[
+    type[BaseException] | None, BeforeValidator(_resolve_exception_class)
+]
+"""An exception class, e.g. ``ValueError``, or the ``module:qualname`` a JSON dump wrote.
+
+Declared here rather than inline because ``TaskFailedEvent`` has a field named ``type``,
+which shadows the builtin for the rest of that class body: an inline
+``type[BaseException]`` after that field is bound raises ``TypeError`` at import, and mypy
+rejects it as "Variable ... is not valid as a type".
+
+``| None`` is inside the alias rather than on the field so the validator runs once, on
+the whole annotation. Spelling the field ``_ExceptionClass | None`` would make it a union
+whose members are each tried in turn, so a ``None`` returned by the validator would
+depend on union resolution order.
+"""
+
+
 def _set_task_fingerprint(event: BaseEvent, task: Any) -> None:
     """Set task identity and fingerprint data on an event."""
     if task is None:
@@ -49,6 +138,18 @@ class TaskFailedEvent(BaseEvent):
     """Event emitted when a task fails"""
 
     error: str
+    error_type: _ExceptionClass = None
+    """The exception's class, e.g. ``ValidationError``.
+
+    The class and not its name, matching ``Telemetry._safe_error_type``: a message
+    is never a type, so ``str(error)`` cannot be passed here at all. A name would
+    not be safe on its own, because a single-word message such as
+    ``"secret_token"`` is itself a valid identifier.
+
+    Kept separate from ``error`` so telemetry can record what kind of failure
+    occurred without ever touching the message, which routinely contains prompts,
+    model output, file paths or credentials.
+    """
     type: Literal["task_failed"] = "task_failed"
     task: Any | None = None
 
@@ -56,6 +157,26 @@ class TaskFailedEvent(BaseEvent):
         super().__init__(**data)
         _set_task_fingerprint(self, self.task)
 
+    @field_serializer("error_type", when_used="json")
+    def _serialize_error_type(self, error_type: _ExceptionClass) -> str | None:
+        """``module:qualname``, so the event stays JSON-serializable and restores exactly.
+
+        A class is not a JSON type, so without this ``model_dump(mode="json")``
+        raises ``PydanticSerializationError`` for the whole event -- which breaks
+        checkpointing after a task failure and sends a ``repr`` to AMP.
+
+        Qualified rather than the bare ``__name__``: a name alone cannot distinguish two
+        exception classes that share one, so restoring from it could pick the wrong
+        class. See ``_resolve_exception_class``.
+
+        ``when_used="json"`` is load-bearing: ``event_listener`` hands the live class
+        to ``Telemetry.task_failed``, which needs it for ``_safe_error_type``, and
+        python-mode dumps must keep it too.
+        """
+        if error_type is None:
+            return None
+        return f"{error_type.__module__}:{error_type.__qualname__}"
+
 
 class TaskEvaluationEvent(BaseEvent):
     """Event emitted when a task evaluation is completed"""

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), 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), task=self),
+            )
             raise e
         finally:
             clear_task_files(self.id)

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

@@ -55,6 +55,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
@@ -608,6 +609,36 @@ class Telemetry:
 
         self._safe_telemetry_operation(_operation)
 
+    def task_failed(
+        self, span: Span, task: Task, error_type: type[BaseException] | None = None
+    ) -> None:
+        """Records that a task execution failed and closes its span with ERROR.
+
+        Failures were previously routed through ``task_ended``, which closes every
+        span as OK - making failed and successful tasks indistinguishable
+        downstream and leaving ``error_count`` at zero for every month on record.
+
+        Takes no ``crew``: unlike ``task_ended`` it reads nothing off one, and
+        requiring it is what caused a task failing without a crew to leak its span
+        instead of being closed.
+
+        Args:
+            span: The OpenTelemetry span tracking the task execution.
+            task: The task that failed.
+            error_type: The exception's class, not its name and never the
+                message. Passed through :meth:`_safe_error_type`, which is why this
+                takes a class: a single-word message like "secret_token" would pass
+                an identifier check, but it is not a type.
+        """
+
+        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, self._safe_error_type(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.
 

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

@@ -121,3 +121,25 @@ 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*, already derived from a class by
+            ``Telemetry._safe_error_type``. The identifier check here is a second
+            gate on that derived name, not the primary defence: callers pass a
+            class precisely because a one-word message such as "secret_token" is
+            itself a valid identifier and would survive this check alone.
+    """
+    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_task_failure_instrumentation.py

@@ -0,0 +1,715 @@
+"""Failed task executions must be recorded as failures, not as successes.
+
+`close_span()` sets `StatusCode.OK` unconditionally, and `TaskFailedEvent` was routed
+to `Telemetry.task_ended`, which calls it. Every failed task was therefore exported as
+OK, which is why `error_count` is not merely low downstream but exactly zero for every
+month on record -- 240.0M task executions across 13 months in
+`crew_task_executions_daily_target`, `error_count = 0` in all of them.
+
+Scope is the task path only. The crew half of the original change (closed PR #6781) is
+deliberately absent: `crew_execution_span()` returns `None` unless `share_crew=True`, so
+`crew._execution_span` is `None` for nearly every user and a crew-failure handler would
+have exited immediately for the default population.
+"""
+
+from contextlib import contextmanager
+from unittest.mock import Mock, patch
+import sys
+import threading
+
+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 otherwise runs with OTEL_SDK_DISABLED=true, which makes TracerProvider
+    hand out non-recording spans that are never exported -- every assertion here would
+    pass vacuously. Set via monkeypatch rather than relying on the root conftest, which
+    pops the variable on teardown and would 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
+    # duration of the test. If it is collected, its processor shuts down and spans
+    # are silently 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 at all."""
+    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 at all -- the span is
+    still ERROR, but no attribute is written.
+    """
+    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(exporter):
+    from crewai.telemetry.telemetry import Telemetry
+
+    exp, tracer = exporter
+    telemetry = Telemetry()
+    telemetry.ready = True
+
+    telemetry.task_failed(
+        tracer.start_span("Task Execution"), 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_task_failed_event_carries_the_class_and_not_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), task=None)
+
+    assert event.error_type is TimeoutError
+    assert "gpt-4o" not in str(event.error_type)
+
+
+def test_a_message_cannot_be_passed_as_the_error_type_at_all():
+    """The field takes the exception *class*, so a message is rejected structurally.
+
+    An identifier check on a name is not sufficient on its own -- a one-word message
+    such as "secret_token" is a valid identifier -- which is exactly why
+    ``Telemetry._safe_error_type`` takes a class rather than a string. Making the event
+    field a class means pydantic refuses a message before any of our code runs.
+    """
+    import pydantic
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    with pytest.raises(pydantic.ValidationError, match="subclass of BaseException"):
+        TaskFailedEvent(error="boom", error_type="secret_token", task=None)
+
+
+def test_task_failed_discards_anything_that_is_not_an_exception_class(exporter):
+    """A non-exception reaching task_failed records nothing, and still closes as ERROR."""
+    from crewai.telemetry.telemetry import Telemetry
+
+    exp, tracer = exporter
+    telemetry = Telemetry()
+    telemetry.ready = True
+
+    telemetry.task_failed(
+        tracer.start_span("Task Execution"),
+        Mock(fingerprint=None),
+        "secret_token",  # type: ignore[arg-type]  - deliberately wrong, as a caller might
+    )
+
+    span = exp.get_finished_spans()[0]
+    assert span.status.status_code is StatusCode.ERROR
+    assert "error_type" not in (span.attributes or {}), (
+        "an identifier-shaped message must not be recorded just because it parses"
+    )
+
+
+def test_error_type_defaults_to_none_for_backwards_compatibility():
+    """Existing callers that construct the event without error_type must keep working."""
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    assert TaskFailedEvent(error="boom", task=None).error_type is None
+
+
+@pytest.fixture
+def listener_with_a_recording_telemetry():
+    """A real EventListener whose telemetry closes spans for real, but nothing else.
+
+    The handler under test is reached through the real event bus, so the routing is
+    genuinely exercised. Only ``Telemetry`` is substituted, and its two relevant methods
+    keep their real span-closing behaviour via ``close_span``/``close_span_with_error``.
+
+    Substituted rather than used live because ``_safe_telemetry_operation`` swallows every
+    exception and returns None: with the real Telemetry, anything that makes it bail --
+    a readiness flag, an env var, a provider already installed by another test in the
+    session -- produces an unclosed span and a failure that looks exactly like the
+    regression this test exists to catch. Verified: the failure moved between the two
+    parameter cases depending only on which ran first in the process.
+
+    Both singletons and the bus are reset on the way in and out, so no handler leaks
+    into another test -- the suite runs in random order.
+    """
+    from crewai.events.event_bus import crewai_event_bus
+    from crewai.events.event_listener import EventListener
+    from crewai.telemetry import Telemetry
+
+    def _reset():
+        with crewai_event_bus._rwlock.w_locked():
+            crewai_event_bus._sync_handlers.clear()
+            crewai_event_bus._async_handlers.clear()
+        Telemetry._instance = None
+        EventListener._instance = None
+        if hasattr(Telemetry, "_lock"):
+            Telemetry._lock = threading.Lock()
+
+    _reset()
+    listener = EventListener()
+
+    class _RecordingTelemetry:
+        def __init__(self):
+            self.calls = []
+
+        def task_failed(self, span, task, error_type=None):
+            self.calls.append("task_failed")
+            # Mirrors the real method: the class is reduced to a safe name first.
+            close_span_with_error(span, Telemetry._safe_error_type(error_type))
+
+        def task_ended(self, span, task, crew):
+            self.calls.append("task_ended")
+            close_span(span)
+
+        def __getattr__(self, _name):
+            return lambda *a, **k: None
+
+    recording = _RecordingTelemetry()
+    listener._telemetry = recording
+    yield listener, crewai_event_bus, recording
+
+    _reset()
+
+
+def _handler_for(bus, event_type):
+    """The handler `setup_listeners` actually registered for an event type.
+
+    Called directly rather than through ``bus.emit``. ``emit`` carries its own
+    event-context and runtime-scope state which is reset per test by an autouse fixture
+    in the root conftest, and with an empty scope stack it does not reliably dispatch --
+    observed here as the same assertion passing or failing depending only on which
+    parametrized case ran first in the process. That machinery is not what this change
+    touches. This still exercises the real closure built by ``setup_listeners``, bound to
+    the real listener, so the routing under test is genuine; it just does not also depend
+    on the emitter.
+    """
+    handlers = list(bus._sync_handlers.get(event_type, []))
+    assert len(handlers) == 1, (
+        f"expected exactly one registered handler for {event_type.__name__}, "
+        f"got {[h.__qualname__ for h in handlers]}"
+    )
+    return handlers[0]
+
+
+def _stub_task(crew):
+    task = Mock()
+    task.name = "some task"
+    task.fingerprint = None
+    task.agent = Mock(crew=crew, role="some role")
+    task.output = None
+    return task
+
+
+@pytest.mark.parametrize(
+    "crew", [Mock(share_crew=False), None], ids=["with-crew", "without-crew"]
+)
+def test_on_task_failed_closes_the_span_as_error_either_way(
+    listener_with_a_recording_telemetry, exporter, crew
+):
+    """Wiring-level coverage, including the case that used to leak the span entirely.
+
+    Two defects on one line. The handler routed to ``task_ended``, which closes the span
+    as OK -- so a failed task was indistinguishable from a successful one downstream.
+    And it only did so when ``source.agent.crew`` was present, so a task failing without
+    one was popped from the span map and then never closed: never ended, never exported,
+    invisible rather than merely mislabelled.
+
+    This is the wiring test CodeRabbit asked for on the original PR and never got, which
+    is why it is here and not only at the ``Telemetry.task_failed`` level.
+    """
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    listener, bus, recording = listener_with_a_recording_telemetry
+    exp, tracer = exporter
+
+    task = _stub_task(crew)
+    listener.execution_spans[task] = tracer.start_span("Task Execution")
+
+    handler = _handler_for(bus, TaskFailedEvent)
+    handler(task, TaskFailedEvent(error="boom", error_type=ValueError, task=task))
+
+    assert recording.calls == ["task_failed"], (
+        "a failure must route to task_failed; task_ended would close the span as OK"
+    )
+    finished = exp.get_finished_spans()
+    assert len(finished) == 1, "the span was never ended, so it would never be exported"
+    assert finished[0].status.status_code is StatusCode.ERROR
+    assert finished[0].attributes["error_type"] == "ValueError"
+    assert task not in listener.execution_spans, "the span map leaked an entry"
+
+
+def test_on_task_completed_still_routes_to_task_ended_and_closes_as_ok(
+    listener_with_a_recording_telemetry, exporter
+):
+    """The success path shares this handler's span map and must be unaffected.
+
+    ``on_task_failed`` and ``on_task_completed`` both pop from ``execution_spans``; only
+    the failure route changed. If this regressed, every successful task would start
+    reporting as an error and error_count would swing from zero to everything -- which is
+    just as wrong and much harder to notice.
+    """
+    from crewai.events.types.task_events import TaskCompletedEvent
+    from crewai.tasks.task_output import TaskOutput
+
+    listener, bus, recording = listener_with_a_recording_telemetry
+    exp, tracer = exporter
+
+    task = _stub_task(Mock(share_crew=False))
+    listener.execution_spans[task] = tracer.start_span("Task Execution")
+
+    output = TaskOutput(description="some task", raw="done", agent="some role")
+    handler = _handler_for(bus, TaskCompletedEvent)
+    handler(task, TaskCompletedEvent(output=output, task=task))
+
+    assert recording.calls == ["task_ended"]
+    finished = exp.get_finished_spans()
+    assert len(finished) == 1
+    assert finished[0].status.status_code is StatusCode.OK
+    assert "error_type" not in (finished[0].attributes or {})
+
+
+class _ProducerFailure(Exception):
+    """Distinctive class, so the test cannot pass on a hardcoded or defaulted value."""
+
+
+@contextmanager
+def captured_task_failures():
+    """Every TaskFailedEvent the producer *emits*, captured at the emit boundary.
+
+    Patches ``crewai_event_bus.emit`` rather than subscribing a handler to it. What these
+    tests are about is what the producer in ``task.py`` constructs, and a subscriber makes
+    that claim depend on the bus choosing to dispatch -- which it does not always do.
+    ``task_failed`` is an "ending" event, and with an empty scope stack (there is no real
+    kickoff here) dispatch is conditional on event-context state that other tests in the
+    same worker process can leave behind. That is not a hypothetical: subscribing passed
+    this file in isolation and every randomized local run, then failed in CI inside a
+    621-test shard with zero events captured.
+
+    The emitted return value is unused by both producers, so returning None is faithful.
+    """
+    from crewai.events.event_bus import crewai_event_bus
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    captured = []
+
+    def _record(_source, event):
+        if isinstance(event, TaskFailedEvent):
+            captured.append(event)
+        return None
+
+    with patch.object(crewai_event_bus, "emit", side_effect=_record):
+        yield captured
+
+
+@pytest.fixture
+def failing_task():
+    """A real Task whose agent raises, so the producer's except block is genuinely run."""
+    from crewai import Agent, Task
+
+    agent = Agent(
+        role="tester",
+        goal="fail",
+        backstory="exists only to raise",
+    )
+    task = Task(description="a task that fails", expected_output="nothing", agent=agent)
+    return task, agent
+
+
+def test_sync_producer_puts_the_exception_class_on_the_event(failing_task):
+    """`Task._execute_core` must populate error_type, not just `error`.
+
+    The tests above construct TaskFailedEvent directly, so a regression in the producer
+    itself -- the two emit sites in task.py -- would pass all of them. This drives the
+    real execution path instead.
+    """
+    from crewai import Agent
+
+    task, _agent = failing_task
+
+    with captured_task_failures() as captured:
+        with patch.object(Agent, "execute_task", side_effect=_ProducerFailure("boom")):
+            with pytest.raises(_ProducerFailure):
+                task._execute_core(None, None, None)
+
+    assert len(captured) == 1, "the producer must emit exactly one TaskFailedEvent"
+    assert captured[0].error_type is _ProducerFailure
+    assert captured[0].error == "boom"
+
+
+@pytest.mark.asyncio
+async def test_async_producer_puts_the_exception_class_on_the_event(failing_task):
+    """The async producer is a separate emit site and regresses independently."""
+    from crewai import Agent
+
+    task, _agent = failing_task
+
+    # aexecute_task, not execute_task: the async producer calls a different agent
+    # method, which is precisely why it can regress independently of the sync one.
+    with captured_task_failures() as captured:
+        with patch.object(Agent, "aexecute_task", side_effect=_ProducerFailure("boom")):
+            with pytest.raises(_ProducerFailure):
+                await task._aexecute_core(None, None, None)
+
+    assert len(captured) == 1
+    assert captured[0].error_type is _ProducerFailure
+    assert captured[0].error == "boom"
+
+
+def test_a_task_with_no_agent_still_reports_a_failure_type(failing_task):
+    """The no-agent branch raises inside the same try, so it must report too."""
+    from crewai import Task
+
+    task = Task(description="orphan", expected_output="nothing")
+
+    with captured_task_failures() as captured:
+        with pytest.raises(Exception, match="has no agent assigned"):
+            task._execute_core(None, None, None)
+
+    assert len(captured) == 1
+    assert captured[0].error_type is Exception
+
+
+def test_the_event_stays_json_serializable_with_a_class_valued_error_type():
+    """A class is not a JSON type, so the field needs a serializer or the event breaks.
+
+    Without one, `model_dump(mode="json")` raises PydanticSerializationError for the
+    *whole* event, not just this field. Two real consumers depend on it: the checkpoint
+    listener dumps every event through EventRecord, and the tracing listener JSON-POSTs
+    events to AMP. So a task failure would take out checkpointing entirely.
+    """
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    event = TaskFailedEvent(error="boom", error_type=ValueError, task=None)
+
+    assert event.model_dump(mode="json")["error_type"] == "builtins:ValueError"
+    assert "ValueError" in event.model_dump_json()
+
+
+def test_python_mode_keeps_the_live_class_for_telemetry():
+    """`when_used="json"` is load-bearing and must not be widened to every mode.
+
+    `event_listener` hands `event.error_type` to `Telemetry.task_failed`, which runs it
+    through `_safe_error_type` -- that requires the class object, not its name. If the
+    serializer applied in python mode too, telemetry would receive a string, silently
+    fail `isinstance(error_type, type)`, and record nothing.
+    """
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    event = TaskFailedEvent(error="boom", error_type=ValueError, task=None)
+
+    assert event.model_dump(mode="python")["error_type"] is ValueError
+    assert event.error_type is ValueError
+
+
+def test_an_absent_error_type_serializes_as_null_not_as_a_string():
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    event = TaskFailedEvent(error="boom", task=None)
+
+    assert event.model_dump(mode="json")["error_type"] is None
+
+
+def test_a_dumped_event_restores_as_itself_and_keeps_both_error_fields():
+    """The JSON dump must round-trip, or restoring a checkpoint loses the failure.
+
+    `_resolve_event` in state/event_record.py wraps `cls.model_validate` in a bare
+    `except Exception` and falls back to `BaseEvent`. So a class-name string the field
+    would not accept does not raise -- it silently degrades the whole event, taking
+    `error` with it even though `error` is a plain string that would have survived.
+    That is worse than the raise this serializer was added to prevent.
+    """
+    from crewai.events.types.task_events import TaskFailedEvent
+    from crewai.state.event_record import EventRecord
+
+    event = TaskFailedEvent(error="boom", error_type=ValueError, task=None)
+    record = EventRecord()
+    record.add(event)
+
+    restored = EventRecord.model_validate(record.model_dump(mode="json")).nodes[
+        event.event_id
+    ].event
+
+    assert type(restored).__name__ == "TaskFailedEvent", (
+        "the event degraded to a bare BaseEvent, so the whole failure was lost"
+    )
+    assert restored.error == "boom"
+    assert restored.error_type is ValueError
+
+
+def test_resolving_a_name_cannot_smuggle_in_a_message():
+    """Accepting the serialized name must not reopen the hole the class type closes.
+
+    Resolution is against real exception classes only, so an identifier-shaped message
+    resolves to nothing and is then rejected by the field's own type.
+    """
+    import pydantic
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    for not_an_exception in ("secret_token", "sk_live_1234", "dict", "os"):
+        with pytest.raises(pydantic.ValidationError, match="subclass of BaseException"):
+            TaskFailedEvent(error="boom", error_type=not_an_exception, task=None)
+
+
+def test_a_non_builtin_exception_class_also_round_trips():
+    """Most failures here are not builtins -- provider and pydantic errors dominate."""
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    event = TaskFailedEvent(error="boom", error_type=_ProducerFailure, task=None)
+    dumped = event.model_dump(mode="json")
+
+    assert dumped["error_type"] == f"{__name__}:_ProducerFailure"
+    assert TaskFailedEvent.model_validate(dumped).error_type is _ProducerFailure
+
+
+def test_two_exception_classes_sharing_a_name_cannot_be_confused():
+    """A same-named class must never be substituted for the one that was serialized.
+
+    Regression for the review finding on this PR: resolving by bare ``__name__`` through
+    ``BaseException.__subclasses__()`` returned whichever same-named class the walk
+    reached first, so an event serialized with one ``DupErr`` restored as a different
+    ``DupErr``. Both are created here exactly as the review reproduced it.
+    """
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    first = type("DupErr", (Exception,), {})
+    second = type("DupErr", (Exception,), {})
+    assert first is not second and first.__name__ == second.__name__
+
+    # Reachable by attribute lookup, so resolution can succeed for the right one.
+    first.__module__ = __name__
+    first.__qualname__ = "_dup_first"
+    second.__module__ = __name__
+    second.__qualname__ = "_dup_second"
+    globals()["_dup_first"] = first
+    globals()["_dup_second"] = second
+    try:
+        dumped = TaskFailedEvent(
+            error="boom", error_type=first, task=None
+        ).model_dump(mode="json")
+
+        restored = TaskFailedEvent.model_validate(dumped).error_type
+
+        assert restored is first
+        assert restored is not second, (
+            "restored a different class that merely shares the name"
+        )
+    finally:
+        del globals()["_dup_first"]
+        del globals()["_dup_second"]
+
+
+def test_a_class_the_running_process_cannot_reach_degrades_to_none_not_to_the_wrong_class():
+    """An unresolvable identity must lose only ``error_type``, never bind to something else.
+
+    A locally defined class is unreachable by attribute lookup -- its ``__qualname__``
+    contains ``<locals>`` -- which is the same shape as a checkpoint written by a process
+    that had a module this one does not. The event must still restore, keeping ``error``.
+    """
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    class LocallyDefined(Exception):
+        pass
+
+    dumped = TaskFailedEvent(
+        error="boom", error_type=LocallyDefined, task=None
+    ).model_dump(mode="json")
+    assert "<locals>" in dumped["error_type"]
+
+    restored = TaskFailedEvent.model_validate(dumped)
+
+    assert restored.error_type is None
+    assert restored.error == "boom", "the message must survive an unresolvable type"
+
+
+def test_a_qualified_identity_cannot_name_a_non_exception_or_force_an_import():
+    """Resolution is gated on being an exception class, and never imports.
+
+    ``os:getcwd`` resolves to a function, and ``json:JSONDecodeError`` is an exception
+    but reachable only if ``json`` is already loaded -- neither may be recorded, and no
+    module absent from ``sys.modules`` may be imported to satisfy a lookup.
+    """
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    unloaded = "xml.dom.minidom"
+    # Restored in `finally`: leaving sys.modules mutated would make this suite
+    # order-dependent, and the runner shuffles tests.
+    previous = sys.modules.pop(unloaded, None)
+    try:
+        for identity in (
+            "os:getcwd",
+            "os:path",
+            f"{unloaded}:Node",
+            "nonexistent_mod:Boom",
+        ):
+            event = TaskFailedEvent(error="boom", error_type=identity, task=None)
+            assert event.error_type is None, identity
+
+        assert unloaded not in sys.modules, "resolution imported a module"
+    finally:
+        if previous is not None:
+            sys.modules[unloaded] = previous
+
+
+def test_a_message_containing_a_colon_is_never_stored():
+    """Messages routinely contain colons, and must not be mistaken for an identity."""
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    for message in (
+        "AuthenticationError: invalid api key sk_live_1234",
+        "connection refused: 10.0.0.1:5432",
+        "secret_token:hunter2",
+    ):
+        event = TaskFailedEvent(error=message, error_type=message, task=None)
+        assert event.error_type is None, message
+
+
+def test_resolution_does_not_trigger_a_lazy_module_getattr():
+    """A PEP 562 ``__getattr__`` must never run: it is an import in disguise.
+
+    Review finding on this PR. ``getattr(module, name)`` invokes a module-level
+    ``__getattr__`` on any miss, and such a hook typically calls
+    ``importlib.import_module`` -- ``crewai.events`` itself is one. So an attribute walk
+    would import a submodule named in the serialized string and run its top-level code,
+    which is exactly the import primitive this resolver must not be. Reading ``__dict__``
+    consults no hook.
+    """
+    import types
+
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    invoked: list[str] = []
+    lazy = types.ModuleType("lazy_probe_pkg")
+
+    def _lazy_getattr(name: str) -> object:
+        invoked.append(name)  # stands in for importlib.import_module
+        raise AttributeError(name)
+
+    lazy.__getattr__ = _lazy_getattr  # type: ignore[attr-defined]
+    sys.modules["lazy_probe_pkg"] = lazy
+    try:
+        event = TaskFailedEvent(
+            error="boom", error_type="lazy_probe_pkg:Something", task=None
+        )
+    finally:
+        del sys.modules["lazy_probe_pkg"]
+
+    assert invoked == [], f"resolution ran the module's __getattr__: {invoked}"
+    assert event.error_type is None
+
+
+def test_a_raising_lazy_hook_cannot_degrade_the_event():
+    """A hook raising anything but AttributeError must not escape validation.
+
+    ``getattr(obj, name, None)`` only swallows ``AttributeError``. Anything else would
+    propagate out of the validator, and ``_resolve_event`` would catch it and degrade the
+    whole event to a bare ``BaseEvent`` -- dropping ``error`` again, which is the
+    regression this resolver was rewritten to prevent.
+    """
+    import types
+
+    from crewai.events.types.task_events import TaskFailedEvent
+
+    exploding = types.ModuleType("exploding_probe_pkg")
+
+    def _exploding_getattr(name: str) -> object:
+        raise RuntimeError("lazy loader exploded")
+
+    exploding.__getattr__ = _exploding_getattr  # type: ignore[attr-defined]
+    sys.modules["exploding_probe_pkg"] = exploding
+    try:
+        event = TaskFailedEvent(
+            error="boom", error_type="exploding_probe_pkg:Something", task=None
+        )
+    finally:
+        del sys.modules["exploding_probe_pkg"]
+
+    assert event.error_type is None
+    assert event.error == "boom", "the message must survive a hostile lazy hook"