feat(flow): report flow outcome, duration and human-in-the-loop signals (#6961)

* feat(flow): report flow outcome and human-in-the-loop signals

A flow reported only that it started. FlowFinishedEvent, FlowFailedEvent,
MethodExecutionFailedEvent, MethodExecutionPausedEvent and FlowPausedEvent all
reached the console formatter and stopped there, and FlowInputRequestedEvent,
FlowInputReceivedEvent and ConversationTurnFailedEvent had no listener at all -
so success rate, failure rate and every HITL pause were unmeasurable.

Adds flow:completed, flow:failed, flow:method_failed, flow:paused,
flow:hitl_paused, flow:input_requested, flow:input_received and
flow:conversation_turn_failed as feature-usage spans, which the existing
feature-usage aggregation already reads.

Deliberately does not hold the Flow Execution span open to measure duration:
flow_executions_daily_target counts those spans at start, so a run that never
finishes would disappear from the count entirely. Duration needs its own span.

Counts only - flow names, method names, error text and flow state are never
recorded.

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

* feat(flow): record how long a flow ran

Adds a Flow Completed span carrying flow_name, duration_ms and outcome,
emitted when a flow finishes or fails. Elapsed time comes from a monotonic
stamp taken at flow start and cleared on use.

Kept separate from the Flow Execution span rather than holding that one open:
it is emitted and closed at start and the daily aggregate counts it, so
holding it would drop every run that is killed or crashes from the execution
count. A killed run now simply has no Flow Completed row, and the count is
unaffected.

Elapsed time is an explicit duration_ms attribute rather than the span's own
duration, which the ingestion pipeline stores as a suffixed string
("0.0000184s") that downstream aggregation parses to zero.

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

* feat(flow): tag flow origin and report resumed runs

Two gaps found while testing the pause/resume path end to end.

Resumed runs were invisible. There is no resume event: a restored run re-enters
through kickoff(), so it looked identical to a fresh start. flow:resumed is
derived from _is_execution_resuming at flow start, which makes
flow:paused - flow:resumed the abandonment rate.

Flow counts are dominated by CrewAI's own AgentExecutor, which is itself a Flow
and runs once per agent execution - it is the top flow in the warehouse by a
wide margin. Nothing distinguished it from a user's flows except guessing at the
name. Both Flow Execution and Flow Completed now carry origin: "internal" when
the flow class is defined under crewai.*, "user" otherwise. Tagging only the new
span would have left the existing daily count unsplittable.

Both span methods take origin with a default, so their signatures stay
backward compatible.

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

* fix(flow): scope outcome and resume signals to user flows

Two findings from review, both confirmed against the code.

Outcome features counted CrewAI's own flows. The agent executor, memory
encoding and memory recall are all Flows and all set suppress_flow_events;
they run far more often than anything a user wrote, so flow:completed,
flow:failed and flow:method_failed were mostly bookkeeping. Those three are now
emitted only for flows the caller wrote. Internal outcomes are still recorded
on the Flow Completed span, which carries origin.

flow:resumed counted checkpoint restores. _is_execution_resuming is set both by
from_pending (a human pause) and by a checkpoint restore that never paused for
anyone, so resumes could exceed pauses and the abandonment rate was unusable.
Keyed off _pending_feedback_context instead, which only from_pending sets.

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

* fix(flow): declare internal flows instead of inferring them

Three findings from review, all confirmed against the code.

Gating on suppress_flow_events was wrong. That flag asks for console quiet and
is a public field, so a caller who set it on their own flow silently lost
flow:completed, flow:failed and flow:method_failed.

Deciding origin from the defining module was also wrong. Flow.from_declaration()
returns a Flow typed in crewai.flow.flow, so a caller's declarative flow was
reported as one of CrewAI's own - the inversion this split exists to prevent.

Both had the same root cause: the discriminator was inferred. Flow now declares
is_crewai_internal, set on the agent executor and the memory encoding/recall
flows, and one helper serves both origin and the outcome gate.

A failed conversational session was reported as completed. Its session closes
with FlowFinishedEvent whatever happened, so a failed turn produced
flow:conversation_turn_failed and flow:completed together. The turn failure is
now recorded on the flow and read back when the session finishes.

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

* refactor(flow): report flow lifecycle as spans, not feature usage

Flow start, completion, pause and method failure are lifecycle facts, and the
lifecycle is reported as spans everywhere else. Reporting them through
feature usage put them in a table that aggregates on the feature string alone -
it cannot carry origin, duration or outcome, so those signals could never be
split between a user's flows and the ones CrewAI runs for itself.

Adds Flow Paused and Flow Method Failed spans, and a resumed marker on Flow
Execution so a run restored from a pause is not counted as a second fresh
start. Removes the duplicate feature rows for completed, failed, method_failed,
paused and resumed - every one of those facts is now on a span, with more
attached to it than the feature row ever carried.

Feature usage keeps only genuine adoption signals: flow:hitl_paused,
flow:input_requested, flow:input_received and flow:conversation_turn_failed.

Also clears the conversational turn-failure flag on every terminal path. A turn
that failed without deferred finalization ends via FlowFailedEvent, and the flag
left set there marked the next run on that instance as failed.

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

* test(flow): update the flow_execution_span caller for the resumed argument

Adding the resumed marker changed a signature that tests/utilities/test_events.py
asserts on exactly, and that assertion was not re-run before pushing.

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

* test(flow): make the checkpoint-restore guard actually guard

The test asserted that flow:resumed was absent from feature usage, but that
signal moved onto the Flow Execution span. The assertion could no longer fail,
so a regression that mis-tagged checkpoint restores as resumes would have gone
unnoticed.

Now asserts the resumed attribute, and waits for the handlers: the manual emit
dispatches asynchronously, so the previous shape also read its result before the
listener had run.

Confirmed it discriminates - keying resumed off _is_execution_resuming again
fails it with [('RestoredFlow', True)] == [('RestoredFlow', False)].

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

* fix(telemetry): record the resumed marker as a string

Verified end to end against the live collector and ClickHouse: the pipeline
encodes a boolean attribute as the presence of a vBool key, so false arrives as
the key simply being absent. That is invisible in the schema and easy to read
wrongly - crew_memory is extracted as "the attribute exists" and consequently
reports 1 for 99.8% of crews against a field that defaults to False.

A string leaves nothing to infer. Confirmed in the warehouse: the emitted span
reads resumed = "false".

Adds direct coverage for the attributes each flow span records, including both
resumed values, and resets the Telemetry singleton in the helper so more than
one span method can be exercised per session.

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

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

docs/edge/ar/telemetry.mdx

@@ -61,6 +61,7 @@ os.environ['OTEL_SDK_DISABLED'] = 'true'
 | نعم       | سمات LLM | تشمل: الاسم، model_name، model، top_k، temperature، واسم فئة LLM. كلها بيانات تقنية غير شخصية. |
 | نعم       | محاولة نشر الطاقم باستخدام CLI الخاص بـ CrewAI | تشمل: حقيقة إجراء النشر ومعرّف الطاقم، وما إذا كان يحاول سحب السجلات، لا بيانات أخرى. |
 | نعم       | بيئة التنفيذ | تشمل: مساعد البرمجة بالذكاء الاصطناعي الذي يشغّل العملية إن وُجد (واحد من قائمة ثابتة مثل `claude_code` أو `codex` أو `cursor` أو `unknown`)، ومكان تشغيل العملية (واحد من قائمة ثابتة مثل `ci` أو `container` أو `serverless` أو `interactive`)، و`project_id` من ملف `pyproject.toml` عند ضبطه. يتحقق الاكتشاف فقط مما إذا كانت متغيرات البيئة المعروفة مضبوطة، ولا يقرأ قيمها أبدًا. لا بيانات شخصية. |
+| نعم       | إشارات دورة حياة التدفق | تشمل: بدء التدفق، وما إذا اكتمل أو فشل، وما إذا فشلت إحدى دواله، وما إذا توقّف مؤقتًا لطلب إدخال أو ملاحظات بشرية، وما إذا كان البدء استئنافًا لتشغيل سابق، وما إذا فشلت دورة محادثة، ومدة تشغيل التدفق، وما إذا كان التدفق من التدفقات التي يشغّلها CrewAI داخليًا أم من كتابتك. يُسجَّل اسم التدفق، كما هو الحال بالفعل عند إنشاء التدفق وتنفيذه. لا تُسجَّل أبدًا أسماء الدوال أو رسائل الأخطاء أو حالة التدفق. لا بيانات شخصية. |
 | لا        | بيانات الوكيل الموسّعة | تشمل: وصف الهدف، نص الخلفية، معرّف ملف موجهات i18n. يجب على المستخدمين التأكد من عدم تضمين معلومات شخصية في حقول النص. |
 | لا        | معلومات المهمة التفصيلية | تشمل: وصف المهمة، وصف المخرجات المتوقعة، مراجع السياق. يجب على المستخدمين التأكد من عدم تضمين معلومات شخصية في هذه الحقول. |
 | لا        | معلومات البيئة | تشمل: المنصة، الإصدار، النظام، الإصدار، وعدد وحدات المعالجة المركزية. مثال: 'Windows 10'، 'x86_64'. لا بيانات شخصية. |

docs/edge/en/telemetry.mdx

@@ -61,6 +61,7 @@ own tracer provider, which is independent of the one described here.
 | Yes       | LLM Attributes | Includes: name, model_name, model, top_k, temperature, and class name of the LLM. All technical, non-personal data. |
 | Yes       | Crew Deployment attempt using crewAI CLI | Includes: The fact a deploy is being made and crew id, and if it's trying to pull logs, no other data. |
 | Yes       | Execution Environment | Includes: which AI coding assistant is running the process, if any (one of a fixed list such as `claude_code`, `codex`, `cursor`, or `unknown`), where the process runs (one of a fixed list such as `ci`, `container`, `serverless`, `interactive`), and the `project_id` from your `pyproject.toml` when one is configured. Detection reads only whether known environment variables are set, never their values. No personal data. |
+| Yes       | Flow Lifecycle Signals | Includes: that a flow started, whether it completed or failed, whether one of its methods failed, whether it paused for human input or feedback, whether the start was a resumed run, whether a conversation turn failed, how long the flow ran, and whether the flow is one CrewAI runs internally or one you wrote. The flow name is recorded, as it already is for flow creation and execution. Method names, error messages and flow state are never recorded. No personal data. |
 | No        | Agent's Expanded Data | Includes: goal description, backstory text, i18n prompt file identifier. Users should ensure no personal info is included in text fields. |
 | No        | Detailed Task Information | Includes: task description, expected output description, context references. Users should ensure no personal info is included in these fields. |
 | No        | Environment Information | Includes: platform, release, system, version, and CPU count. Example: 'Windows 10', 'x86_64'. No personal data. |

docs/edge/ko/telemetry.mdx

@@ -59,6 +59,7 @@ provider로 등록하지 않습니다. 이를 통해 양방향이 분리됩니
 | 예      | LLM 속성                                  | LLM의 이름, model_name, 모델, top_k, temperature 및 클래스명이 포함됩니다. 모두 기술적이고 비개인 정보입니다. |
 | 예      | crewAI CLI를 통한 Crew 배포 시도           | 배포가 시도되고 있고 crew id가 포함되며, 로그를 가져오려고 하는 경우에만 해당. 다른 데이터 없음. |
 | 예      | 실행 환경                                  | 포함: 프로세스를 실행 중인 AI 코딩 어시스턴트(있는 경우, `claude_code`, `codex`, `cursor`, `unknown` 등 고정 목록 중 하나), 프로세스가 실행되는 위치(`ci`, `container`, `serverless`, `interactive` 등 고정 목록 중 하나), 그리고 `pyproject.toml`에 설정된 경우 `project_id`. 감지는 알려진 환경 변수의 설정 여부만 확인하며 값은 읽지 않음. 개인 데이터 없음. |
+| 예      | Flow 라이프사이클 신호 | 포함 항목: flow의 시작, 완료 또는 실패 여부, 해당 메서드의 실패 여부, 사람의 입력이나 피드백을 위해 일시 중지되었는지 여부, 해당 시작이 재개된 실행인지 여부, 대화 턴의 실패 여부, flow 실행 시간, 그리고 해당 flow가 CrewAI가 내부적으로 실행하는 것인지 사용자가 작성한 것인지 여부. flow 이름은 기록되며, 이는 flow 생성 및 실행에서 이미 그러합니다. 메서드 이름, 오류 메시지, flow 상태는 절대 기록하지 않습니다. 개인 정보 없음. |
 | 아니오  | 에이전트 확장 데이터                      | 목표 설명, 배경 이야기 텍스트, i18n 프롬프트 파일 식별자가 포함됩니다. 사용자들은 텍스트 필드에 개인 정보가 포함되지 않도록 해야 합니다. |
 | 아니오  | 상세 작업 정보                            | 작업 설명, 예상 출력 설명, 컨텍스트 참조가 포함됩니다. 사용자들은 이러한 필드에 개인 정보가 포함되지 않도록 해야 합니다. |
 | 아니오  | 환경 정보                                 | 플랫폼, 릴리즈, 시스템, 버전, CPU 개수가 포함됩니다. 예: 'Windows 10', 'x86_64'. 개인 정보 없음. |

docs/edge/pt-BR/telemetry.mdx

@@ -61,6 +61,7 @@ por meio do próprio tracer provider, que é independente do descrito aqui.
 | 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    | Tentativa de Deploy do Crew pelo CLI do crewAI | Inclui: O fato de um deploy estar sendo realizado e o crew id, e se está tentando buscar logs, sem mais dados.                          |
 | Sim    | Ambiente de Execução                       | Inclui: qual assistente de código com IA está executando o processo, se houver (um de uma lista fixa como `claude_code`, `codex`, `cursor` ou `unknown`), onde o processo é executado (um de uma lista fixa como `ci`, `container`, `serverless`, `interactive`) e o `project_id` do seu `pyproject.toml` quando houver um configurado. A detecção lê apenas se variáveis de ambiente conhecidas estão definidas, nunca seus valores. Sem dados pessoais. |
+| Sim    | Sinais de Ciclo de Vida do Flow | Inclui: que um flow iniciou, se foi concluído ou falhou, se um de seus métodos falhou, se pausou para entrada ou feedback humano, se o início foi uma execução retomada, se um turno de conversa falhou, por quanto tempo o flow executou e se o flow é um que o CrewAI executa internamente ou um que você escreveu. O nome do flow é registrado, como já ocorre na criação e execução do flow. Nomes de métodos, mensagens de erro e estado do flow nunca são registrados. Sem dados pessoais. |
 | Não    | Dados Expandidos do Agente                 | Inclui: descrição do objetivo, texto da história, identificador de arquivo i18n prompt. Usuários devem garantir que não haja info pessoal nesses campos de texto.            |
 | Não    | Informações Detalhadas da Tarefa           | Inclui: descrição da tarefa, descrição do resultado esperado, referências de contexto. Usuários devem garantir que não haja info pessoal nessas áreas.           |
 | Não    | Informações de Ambiente                    | Inclui: plataforma, release, sistema, versão e quantidade de CPUs. Exemplo: 'Windows 10', 'x86_64'. Sem dados pessoais.                       |

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

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from io import StringIO
+import time
 from typing import TYPE_CHECKING, Any
 
 from pydantic import Field, PrivateAttr
@@ -42,9 +43,12 @@ from crewai.events.types.env_events import (
 )
 from crewai.events.types.flow_events import (
     ConversationTurnCompletedEvent,
+    ConversationTurnFailedEvent,
     FlowCreatedEvent,
     FlowFailedEvent,
     FlowFinishedEvent,
+    FlowInputReceivedEvent,
+    FlowInputRequestedEvent,
     FlowPausedEvent,
     FlowStartedEvent,
     HumanFeedbackReceivedEvent,
@@ -304,17 +308,76 @@ class EventListener(BaseEventListener):
         def on_flow_created(_: Any, event: FlowCreatedEvent) -> None:
             self._telemetry.flow_creation_span(event.flow_name)
 
+        def _flow_origin(source: Any) -> str:
+            """Separate flows CrewAI runs itself from the ones a caller wrote.
+
+            The agent executor and the memory encoding/recall flows are all
+            Flows and run far more often than anything a user wrote, so without
+            this they swamp every flow metric.
+
+            Reads the declared ``is_crewai_internal`` marker rather than the
+            defining module: a declarative flow built with
+            ``Flow.from_declaration()`` is typed as ``Flow`` itself, so a module
+            check would report a caller's flow as internal. It is also not
+            ``suppress_flow_events``, which only asks for console quiet and can
+            legitimately be set on a caller's own flow.
+            """
+            return (
+                "internal"
+                if getattr(type(source), "is_crewai_internal", False)
+                else "user"
+            )
+
+        def _report_flow_duration(source: Any, flow_name: str, outcome: str) -> None:
+            """Emit the elapsed time for a flow that reached a terminal state.
+
+            A flow can finish without this listener having seen it start - a
+            conversational turn re-emits completion for a restored run - so a
+            missing stamp means "no duration to report", not an error.
+            """
+            # Cleared on every terminal path, not only on finish: a turn that
+            # fails without deferred finalization ends via FlowFailedEvent, and
+            # a flag left set there would mark the next run on this instance
+            # failed.
+            source._telemetry_turn_failed = False
+            started_at = getattr(source, "_telemetry_started_at", None)
+            if started_at is None:
+                return
+            source._telemetry_started_at = None
+            self._telemetry.flow_completed_span(
+                flow_name,
+                (time.monotonic() - started_at) * 1000,
+                outcome,
+                _flow_origin(source),
+            )
+
         @crewai_event_bus.on(FlowStartedEvent)
         def on_flow_started(source: Any, event: FlowStartedEvent) -> None:
+            # A run restored from a pause is only visible here: there is no
+            # resume event, and resuming re-enters kickoff(). Keyed off the
+            # pending-feedback context rather than _is_execution_resuming, which
+            # a checkpoint restore also sets even though nobody ever paused.
+            resumed = getattr(source, "_pending_feedback_context", None) is not None
             self._telemetry.flow_execution_span(
-                event.flow_name, list(source._methods.keys())
+                event.flow_name,
+                list(source._methods.keys()),
+                _flow_origin(source),
+                resumed,
             )
+            source._telemetry_started_at = time.monotonic()
             if not getattr(source, "suppress_flow_events", False):
                 self.formatter.handle_flow_created(event.flow_name, str(source.flow_id))
                 self.formatter.handle_flow_started(event.flow_name, str(source.flow_id))
 
         @crewai_event_bus.on(FlowFinishedEvent)
         def on_flow_finished(source: Any, event: FlowFinishedEvent) -> None:
+            outcome = (
+                "failed"
+                if getattr(source, "_telemetry_turn_failed", False)
+                else "completed"
+            )
+            _report_flow_duration(source, event.flow_name, outcome)
+
             if not getattr(source, "suppress_flow_events", False):
                 self.formatter.handle_flow_status(
                     event.flow_name,
@@ -323,6 +386,8 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(FlowFailedEvent)
         def on_flow_failed(source: Any, event: FlowFailedEvent) -> None:
+            _report_flow_duration(source, event.flow_name, "failed")
+
             if not getattr(source, "suppress_flow_events", False):
                 self.formatter.handle_flow_status(
                     event.flow_name,
@@ -336,6 +401,23 @@ class EventListener(BaseEventListener):
         ) -> None:
             self._telemetry.feature_usage_span("flow:conversation_turn")
 
+        @crewai_event_bus.on(ConversationTurnFailedEvent)
+        def on_conversation_turn_failed(
+            source: Any, event: ConversationTurnFailedEvent
+        ) -> None:
+            self._telemetry.feature_usage_span("flow:conversation_turn_failed")
+            # A conversational session closes with FlowFinishedEvent whatever
+            # happened, so record the failure for on_flow_finished to read.
+            source._telemetry_turn_failed = True
+
+        @crewai_event_bus.on(FlowInputRequestedEvent)
+        def on_flow_input_requested(_: Any, event: FlowInputRequestedEvent) -> None:
+            self._telemetry.feature_usage_span("flow:input_requested")
+
+        @crewai_event_bus.on(FlowInputReceivedEvent)
+        def on_flow_input_received(_: Any, event: FlowInputReceivedEvent) -> None:
+            self._telemetry.feature_usage_span("flow:input_received")
+
         @crewai_event_bus.on(MethodExecutionStartedEvent)
         def on_method_execution_started(
             source: Any, event: MethodExecutionStartedEvent
@@ -360,8 +442,14 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(MethodExecutionFailedEvent)
         def on_method_execution_failed(
-            _: Any, event: MethodExecutionFailedEvent
+            source: Any, event: MethodExecutionFailedEvent
         ) -> None:
+            # The method name is not recorded: it is user-authored and would put
+            # arbitrary strings in telemetry.
+            self._telemetry.flow_method_failed_span(
+                event.flow_name, _flow_origin(source)
+            )
+
             self.formatter.handle_method_status(
                 event.method_name,
                 "failed",
@@ -371,13 +459,17 @@ class EventListener(BaseEventListener):
         def on_method_execution_paused(
             _: Any, event: MethodExecutionPausedEvent
         ) -> None:
+            self._telemetry.feature_usage_span("flow:hitl_paused")
+
             self.formatter.handle_method_status(
                 event.method_name,
                 "paused",
             )
 
         @crewai_event_bus.on(FlowPausedEvent)
-        def on_flow_paused(_: Any, event: FlowPausedEvent) -> None:
+        def on_flow_paused(source: Any, event: FlowPausedEvent) -> None:
+            self._telemetry.flow_paused_span(event.flow_name, _flow_origin(source))
+
             self.formatter.handle_flow_status(
                 event.flow_name,
                 event.flow_id,

lib/crewai/src/crewai/experimental/agent_executor.py

@@ -9,7 +9,7 @@ from datetime import datetime
 import inspect
 import json
 import threading
-from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast
+from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypeVar, cast
 from uuid import uuid4
 
 from crewai_core.printer import PRINTER
@@ -189,6 +189,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
 
     executor_type: Literal["experimental"] = "experimental"
     suppress_flow_events: bool = True  # always suppress for executor
+    is_crewai_internal: ClassVar[bool] = True
     llm: BaseLLM | None = Field(default=None, exclude=True)
     prompt: SystemPromptResult | StandardPromptResult | None = Field(
         default=None, exclude=True

lib/crewai/src/crewai/flow/runtime/__init__.py

@@ -717,8 +717,24 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
     _or_listeners_lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)
     _completed_methods: set[FlowMethodName] = PrivateAttr(default_factory=set)
     _method_call_counts: dict[FlowMethodName, int] = PrivateAttr(default_factory=dict)
+    # True on flows CrewAI runs for its own bookkeeping - the agent executor and
+    # the memory encoding/recall flows. Declared rather than inferred from the
+    # module: a declarative flow built with Flow.from_declaration() is typed as
+    # Flow itself, so a module check would call a caller's flow internal.
+    is_crewai_internal: ClassVar[bool] = False
+
+    # Set by the telemetry listener when a conversational turn fails. A
+    # conversational session emits FlowFinishedEvent from
+    # finalize_session_traces() regardless of outcome, so without this a failed
+    # session would be reported as a successful completion.
+    _telemetry_turn_failed: bool = PrivateAttr(default=False)
+
     _is_execution_resuming: bool = PrivateAttr(default=False)
     _restored_from_checkpoint: bool = PrivateAttr(default=False)
+    # Monotonic stamp set by the telemetry listener at flow start, so the
+    # duration span emitted at the end does not need to hold a span open for
+    # the life of the run.
+    _telemetry_started_at: float | None = PrivateAttr(default=None)
     _event_futures: list[Future[None]] = PrivateAttr(default_factory=list)
     _pending_feedback_context: PendingFeedbackContext | None = PrivateAttr(default=None)
     _human_feedback_method_outputs: dict[str, Any] = PrivateAttr(default_factory=dict)

lib/crewai/src/crewai/memory/encoding_flow.py

@@ -15,7 +15,7 @@ import contextvars
 from datetime import datetime
 import logging
 import math
-from typing import Any
+from typing import Any, ClassVar
 from uuid import uuid4
 
 from pydantic import BaseModel, Field
@@ -82,6 +82,8 @@ class EncodingFlow(Flow[EncodingState]):
     - ONE batch re-embed for updates + ONE bulk storage write
     """
 
+    is_crewai_internal: ClassVar[bool] = True
+
     _skip_auto_memory: bool = True
 
     initial_state: type[EncodingState] = EncodingState

lib/crewai/src/crewai/memory/recall_flow.py

@@ -14,7 +14,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 import contextvars
 from datetime import datetime
 import logging
-from typing import Any
+from typing import Any, ClassVar
 from uuid import uuid4
 
 from pydantic import BaseModel, Field
@@ -63,6 +63,8 @@ class RecallFlow(Flow[RecallState]):
     and iteratively deepens exploration when confidence is low.
     """
 
+    is_crewai_internal: ClassVar[bool] = True
+
     _skip_auto_memory: bool = True
 
     initial_state: type[RecallState] = RecallState

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

@@ -991,12 +991,26 @@ class Telemetry:
 
         self._safe_telemetry_operation(_operation)
 
-    def flow_execution_span(self, flow_name: str, node_names: list[str]) -> None:
+    def flow_execution_span(
+        self,
+        flow_name: str,
+        node_names: list[str],
+        origin: str = "user",
+        resumed: bool = False,
+    ) -> None:
         """Records the execution of a flow.
 
         Args:
             flow_name: Name of the flow being executed.
             node_names: List of nodes being executed in the flow.
+            origin: ``"internal"`` for flows CrewAI itself runs (the agent
+                executor), ``"user"`` for flows the caller authored. Without it
+                the agent executor, which runs once per agent execution, is
+                indistinguishable from a user's own flows in the daily counts.
+            resumed: True when this start is a run restored from a human pause.
+                Resuming re-enters ``kickoff()``, so the same event fires again;
+                without this the second leg is indistinguishable from a fresh
+                run and a paused flow looks like two separate executions.
         """
 
         def _operation() -> None:
@@ -1009,6 +1023,94 @@ class Telemetry:
             )
             self._add_attribute(span, "flow_name", flow_name)
             self._add_attribute(span, "node_names", json.dumps(node_names))
+            self._add_attribute(span, "origin", origin)
+            # Recorded as a string rather than a bool. The pipeline encodes a
+            # boolean as the presence of a vBool key - false arrives as the key
+            # simply being absent - which is invisible in the schema and easy to
+            # extract wrongly. crew_memory reads 1 for 99.8% of crews for exactly
+            # that reason, against a field that defaults to False.
+            self._add_attribute(span, "resumed", "true" if resumed else "false")
+            close_span(span)
+
+        self._safe_telemetry_operation(_operation)
+
+    def flow_completed_span(
+        self, flow_name: str, duration_ms: float, outcome: str, origin: str = "user"
+    ) -> None:
+        """Records how long a flow ran and how it ended.
+
+        A separate span from ``Flow Execution`` rather than that span held open
+        to completion: ``Flow Execution`` is emitted and closed at start, and
+        the daily aggregate counts it, so holding it would drop every run that
+        is killed or crashes from the execution count entirely.
+
+        The elapsed time is recorded as an explicit ``duration_ms`` attribute
+        rather than left to the span's own duration, which the ingestion
+        pipeline stores as a suffixed string ("0.0000184s") that downstream
+        aggregation cannot parse.
+
+        Args:
+            flow_name: Name of the flow that finished.
+            duration_ms: Wall-clock milliseconds from flow start, measured on a
+                monotonic clock.
+            outcome: Either ``"completed"`` or ``"failed"``.
+            origin: ``"internal"`` for flows CrewAI itself runs (the agent
+                executor), ``"user"`` for flows the caller authored.
+        """
+
+        def _operation() -> None:
+            tracer = self.provider.get_tracer(TRACER_NAME)
+            span = tracer.start_span("Flow Completed")
+            self._add_attribute(span, "crewai_version", version("crewai"))
+            self._add_attribute(span, "flow_name", flow_name)
+            self._add_attribute(span, "duration_ms", duration_ms)
+            self._add_attribute(span, "outcome", outcome)
+            self._add_attribute(span, "origin", origin)
+            close_span(span)
+
+        self._safe_telemetry_operation(_operation)
+
+    def flow_paused_span(self, flow_name: str, origin: str = "user") -> None:
+        """Records that a flow stopped to wait for a human.
+
+        A pause is a lifecycle state, not a feature: the run has neither
+        completed nor failed, so it appears in neither terminal span. Without
+        this a paused flow is simply a start with no end.
+
+        Args:
+            flow_name: Name of the flow that paused.
+            origin: ``"internal"`` or ``"user"`` - see
+                :meth:`flow_execution_span`.
+        """
+
+        def _operation() -> None:
+            tracer = self.provider.get_tracer(TRACER_NAME)
+            span = tracer.start_span("Flow Paused")
+            self._add_attribute(span, "crewai_version", version("crewai"))
+            self._add_attribute(span, "flow_name", flow_name)
+            self._add_attribute(span, "origin", origin)
+            close_span(span)
+
+        self._safe_telemetry_operation(_operation)
+
+    def flow_method_failed_span(self, flow_name: str, origin: str = "user") -> None:
+        """Records that a method inside a flow raised.
+
+        The method name is deliberately not recorded: it is user-authored and
+        would put arbitrary strings in telemetry.
+
+        Args:
+            flow_name: Name of the flow whose method failed.
+            origin: ``"internal"`` or ``"user"`` - see
+                :meth:`flow_execution_span`.
+        """
+
+        def _operation() -> None:
+            tracer = self.provider.get_tracer(TRACER_NAME)
+            span = tracer.start_span("Flow Method Failed")
+            self._add_attribute(span, "crewai_version", version("crewai"))
+            self._add_attribute(span, "flow_name", flow_name)
+            self._add_attribute(span, "origin", origin)
             close_span(span)
 
         self._safe_telemetry_operation(_operation)

lib/crewai/tests/telemetry/test_flow_telemetry.py

@@ -0,0 +1,617 @@
+"""Flow outcome and human-in-the-loop signals must reach telemetry.
+
+Driven through real ``Flow`` executions rather than by emitting events directly,
+so these fail if the event bus, the listener wiring, or the emitting call site
+changes - not just if the listener body does.
+
+Before this, a flow reported only that it *started*: ``FlowFinishedEvent``,
+``FlowFailedEvent``, ``MethodExecutionFailedEvent``, ``MethodExecutionPausedEvent``
+and ``FlowPausedEvent`` all reached the console formatter and stopped there, and
+the input and conversation-failure events had no listener at all.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import time
+
+import pytest
+
+from crewai.flow.async_feedback import HumanFeedbackPending, PendingFeedbackContext
+from crewai.flow.flow import Flow, listen, start
+from crewai.flow.human_feedback import human_feedback
+from crewai.flow.input_provider import InputResponse
+
+from ..utils import wait_for_event_handlers
+
+
+def _reregister_listener() -> None:
+    """Re-subscribe the global listener to the event bus.
+
+    The repo-wide ``cleanup_event_handlers`` fixture clears every handler after
+    each test, so anything relying on the shared listener sees an empty bus
+    unless it happens to run first.
+    """
+    from crewai.events import event_listener as listener_module
+    from crewai.events.event_bus import crewai_event_bus
+    from crewai.events.types.flow_events import FlowStartedEvent
+
+    # Only when the bus is empty: subscribing a second time registers a fresh
+    # set of closures, and every handler then fires twice.
+    if crewai_event_bus._sync_handlers.get(FlowStartedEvent):
+        return
+
+    listener_module.event_listener.setup_listeners(crewai_event_bus)
+
+
+@pytest.fixture
+def flow_spans(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str]]:
+    """Record (flow_name, origin) for every Flow Execution span."""
+    from crewai.events import event_listener as listener_module
+
+    _reregister_listener()
+
+    recorded: list[tuple[str, str]] = []
+    monkeypatch.setattr(
+        listener_module.event_listener._telemetry,
+        "flow_execution_span",
+        lambda flow_name, node_names, origin="user", resumed=False: recorded.append(
+            (flow_name, origin)
+        ),
+    )
+    return recorded
+
+
+@pytest.fixture
+def starts(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, bool]]:
+    """Record (flow_name, resumed) for every Flow Execution span."""
+    from crewai.events import event_listener as listener_module
+
+    _reregister_listener()
+
+    recorded: list[tuple[str, bool]] = []
+    monkeypatch.setattr(
+        listener_module.event_listener._telemetry,
+        "flow_execution_span",
+        lambda flow_name, node_names, origin="user", resumed=False: recorded.append(
+            (flow_name, resumed)
+        ),
+    )
+    return recorded
+
+
+@pytest.fixture
+def pauses(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str]]:
+    """Record (flow_name, origin) for every Flow Paused span."""
+    from crewai.events import event_listener as listener_module
+
+    _reregister_listener()
+
+    recorded: list[tuple[str, str]] = []
+    monkeypatch.setattr(
+        listener_module.event_listener._telemetry,
+        "flow_paused_span",
+        lambda flow_name, origin="user": recorded.append((flow_name, origin)),
+    )
+    return recorded
+
+
+@pytest.fixture
+def method_failures(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str]]:
+    """Record (flow_name, origin) for every Flow Method Failed span."""
+    from crewai.events import event_listener as listener_module
+
+    _reregister_listener()
+
+    recorded: list[tuple[str, str]] = []
+    monkeypatch.setattr(
+        listener_module.event_listener._telemetry,
+        "flow_method_failed_span",
+        lambda flow_name, origin="user": recorded.append((flow_name, origin)),
+    )
+    return recorded
+
+
+@pytest.fixture
+def durations(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, float, str]]:
+    """Record every (flow_name, duration_ms, outcome) the listener reports."""
+    from crewai.events import event_listener as listener_module
+
+    _reregister_listener()
+
+    recorded: list[tuple[str, float, str]] = []
+    monkeypatch.setattr(
+        listener_module.event_listener._telemetry,
+        "flow_completed_span",
+        lambda flow_name, duration_ms, outcome, origin="user": recorded.append(
+            (flow_name, duration_ms, outcome)
+        ),
+    )
+    return recorded
+
+
+@pytest.fixture
+def features(monkeypatch: pytest.MonkeyPatch) -> list[str]:
+    """Record every feature the listener reports for a real flow run.
+
+    Observes the telemetry boundary rather than exported spans: the suite builds
+    the Telemetry singleton with collection disabled, so it has no provider to
+    export through, and replacing that singleton mid-session leaves the event
+    bus without its handlers. That the recorded features become spans is covered
+    by ``test_tracer_isolation``.
+    """
+    from crewai.events import event_listener as listener_module
+
+    _reregister_listener()
+
+    recorded: list[str] = []
+    monkeypatch.setattr(
+        listener_module.event_listener._telemetry,
+        "feature_usage_span",
+        recorded.append,
+    )
+    return recorded
+
+
+def test_completed_flow_reports_its_outcome(
+    durations: list[tuple[str, float, str]],
+) -> None:
+    """Outcome is a lifecycle fact, so it belongs on a span, not a feature."""
+
+    class OkFlow(Flow):
+        @start()
+        def go(self) -> str:
+            return "ok"
+
+    OkFlow().kickoff()
+
+    assert [(n, o) for n, _d, o in durations] == [("OkFlow", "completed")]
+
+
+def test_failed_flow_reports_the_failure_and_the_method(
+    durations: list[tuple[str, float, str]], method_failures: list[tuple[str, str]]
+) -> None:
+    class BoomFlow(Flow):
+        @start()
+        def go(self) -> str:
+            raise RuntimeError("boom")
+
+    with pytest.raises(RuntimeError, match="boom"):
+        BoomFlow().kickoff()
+
+    assert [(n, o) for n, _d, o in durations] == [("BoomFlow", "failed")]
+    assert ("BoomFlow", "user") in method_failures
+
+
+def test_a_failed_flow_is_still_counted_as_an_execution(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The start-time span must survive, or aborted runs vanish from counts.
+
+    ``flow_executions_daily_target`` counts ``Flow Execution`` spans, emitted
+    when the flow starts. Holding that span open until completion to measure
+    duration - the obvious way to add duration - would drop every run that never
+    finishes, so the outcome signals are reported separately instead.
+    """
+    from crewai.events import event_listener as listener_module
+
+    _reregister_listener()
+
+    started: list[str] = []
+    monkeypatch.setattr(
+        listener_module.event_listener._telemetry,
+        "flow_execution_span",
+        lambda flow_name, node_names, origin="user", resumed=False: started.append(
+            flow_name
+        ),
+    )
+
+    class BoomFlow(Flow):
+        @start()
+        def go(self) -> str:
+            raise RuntimeError("boom")
+
+    with pytest.raises(RuntimeError, match="boom"):
+        BoomFlow().kickoff()
+
+    assert "BoomFlow" in started
+
+
+def test_requesting_input_reports_both_sides(features: list[str]) -> None:
+    class StubProvider:
+        def request_input(self, message: str, flow: Flow, metadata=None):
+            return InputResponse(value="typed answer")
+
+    class AskFlow(Flow):
+        @start()
+        def go(self) -> str:
+            return self.ask("What topic?")
+
+    AskFlow(input_provider=StubProvider()).kickoff()
+
+    emitted = features
+    assert "flow:input_requested" in emitted
+    assert "flow:input_received" in emitted
+
+
+def test_paused_flow_reports_the_pause(
+    features: list[str], pauses: list[tuple[str, str]]
+) -> None:
+    """An async feedback provider pauses the flow; both signals must land."""
+
+    class AsyncProvider:
+        def request_feedback(self, context: PendingFeedbackContext, flow: Flow) -> str:
+            raise HumanFeedbackPending(context=context)
+
+    class PausingFlow(Flow):
+        @start()
+        @human_feedback(message="Review:", provider=AsyncProvider())
+        def generate(self) -> str:
+            return "content"
+
+        @listen(generate)
+        def process(self, result) -> str:
+            return f"processed: {result.feedback}"
+
+    # Whether the pause surfaces as an exception depends on the persistence
+    # backend in use; the signals must land either way.
+    with contextlib.suppress(BaseException):
+        PausingFlow().kickoff()
+
+    # The pause itself is lifecycle and lands on a span; that a human-feedback
+    # method was what paused is genuine feature adoption.
+    assert ("PausingFlow", "user") in pauses
+    assert "flow:hitl_paused" in features
+
+
+def test_failed_conversation_turn_is_reported(features: list[str]) -> None:
+    """Only completed turns were tracked, so failure rate was unknowable."""
+
+    class FailingChat(Flow):
+        conversational = True
+
+        @start()
+        def begin(self) -> str:
+            raise RuntimeError("turn exploded")
+
+    with pytest.raises(RuntimeError, match="turn exploded"):
+        FailingChat().handle_turn("hello")
+
+    assert "flow:conversation_turn_failed" in features
+
+
+def test_no_method_names_or_error_text_are_recorded(
+    method_failures: list[tuple[str, str]],
+    durations: list[tuple[str, float, str]],
+    features: list[str],
+) -> None:
+    """Method names and error text are user-authored and must not be sent.
+
+    The flow name is recorded, as it already is for flow creation and
+    execution, so it is deliberately not asserted against here.
+    """
+
+    class SecretNamedFlow(Flow):
+        @start()
+        def my_secret_method_name(self) -> str:
+            raise RuntimeError("secret error detail")
+
+    with pytest.raises(RuntimeError, match="secret error detail"):
+        SecretNamedFlow().kickoff()
+
+    assert method_failures, "the failure must still be reported"
+    recorded = [
+        str(value)
+        for row in (*method_failures, *durations)
+        for value in row
+    ] + features
+    for value in recorded:
+        assert "my_secret_method_name" not in value
+        assert "secret error detail" not in value
+
+
+def test_completed_flow_reports_a_real_duration(
+    durations: list[tuple[str, float, str]],
+) -> None:
+    """Elapsed time must be measured, not merely present."""
+
+    class SlowFlow(Flow):
+        @start()
+        def go(self) -> str:
+            time.sleep(0.05)
+            return "ok"
+
+    SlowFlow().kickoff()
+
+    assert len(durations) == 1
+    flow_name, duration_ms, outcome = durations[0]
+    assert flow_name == "SlowFlow"
+    assert outcome == "completed"
+    assert duration_ms >= 50
+
+
+def test_failed_flow_reports_its_duration_and_outcome(
+    durations: list[tuple[str, float, str]],
+) -> None:
+    class SlowBoomFlow(Flow):
+        @start()
+        def go(self) -> str:
+            time.sleep(0.05)
+            raise RuntimeError("boom")
+
+    with pytest.raises(RuntimeError, match="boom"):
+        SlowBoomFlow().kickoff()
+
+    assert len(durations) == 1
+    flow_name, duration_ms, outcome = durations[0]
+    assert flow_name == "SlowBoomFlow"
+    assert outcome == "failed"
+    assert duration_ms >= 50
+
+
+def test_no_duration_is_reported_without_a_recorded_start(
+    durations: list[tuple[str, float, str]],
+) -> None:
+    """A completion with no observed start reports nothing, and does not raise.
+
+    A conversational turn can re-emit completion for a restored run, so the
+    stamp is genuinely absent rather than impossible.
+    """
+    from crewai.events.event_bus import crewai_event_bus
+    from crewai.events.types.flow_events import FlowFinishedEvent
+
+    class NeverStartedFlow(Flow):
+        @start()
+        def go(self) -> str:
+            return "ok"
+
+    flow = NeverStartedFlow()
+    crewai_event_bus.emit(
+        flow,
+        FlowFinishedEvent(flow_name="NeverStartedFlow", result="ok", state={}),
+    )
+
+    assert durations == []
+
+
+def test_duration_is_reported_once_per_run(
+    durations: list[tuple[str, float, str]],
+) -> None:
+    """The stamp is cleared on use, so a repeated completion cannot double-count."""
+    from crewai.events.event_bus import crewai_event_bus
+    from crewai.events.types.flow_events import FlowFinishedEvent
+
+    class OkFlow(Flow):
+        @start()
+        def go(self) -> str:
+            return "ok"
+
+    flow = OkFlow()
+    flow.kickoff()
+    crewai_event_bus.emit(
+        flow, FlowFinishedEvent(flow_name="OkFlow", result="ok", state={})
+    )
+
+    assert len(durations) == 1
+
+
+def test_user_authored_flows_are_tagged_as_user(flow_spans) -> None:
+    class MyOwnFlow(Flow):
+        @start()
+        def go(self) -> str:
+            return "ok"
+
+    MyOwnFlow().kickoff()
+
+    assert ("MyOwnFlow", "user") in flow_spans
+
+
+def test_crewais_own_agent_executor_is_tagged_internal(flow_spans) -> None:
+    """The agent executor is a Flow and runs once per agent execution.
+
+    Without an origin tag it is indistinguishable from a user's flows in the
+    daily counts, and it dominates them.
+    """
+    from crewai import Agent, Crew, Task
+    from crewai.llms.base_llm import BaseLLM
+
+    class StubLLM(BaseLLM):
+        def __init__(self) -> None:
+            super().__init__(model="stub-model")
+
+        def call(self, messages, **kwargs) -> str:
+            return "Final Answer: done"
+
+        def supports_function_calling(self) -> bool:
+            return False
+
+        def supports_stop_words(self) -> bool:
+            return False
+
+        def get_context_window_size(self) -> int:
+            return 8192
+
+    agent = Agent(role="R", goal="G", backstory="B", llm=StubLLM())
+    task = Task(description="Do it", expected_output="A result", agent=agent)
+    Crew(agents=[agent], tasks=[task]).kickoff()
+
+    origins = {name: origin for name, origin in flow_spans}
+    assert origins.get("AgentExecutor") == "internal"
+
+
+def test_resumed_flow_is_reported(
+    tmp_path,
+    pauses: list[tuple[str, str]],
+    starts: list[tuple[str, bool]],
+    durations: list[tuple[str, float, str]],
+) -> None:
+    """A restored run is only visible here - there is no resume event.
+
+    Without it, a paused flow that was abandoned cannot be told apart from one
+    the user came back to.
+    """
+    from pydantic import BaseModel
+
+    from crewai.events.event_bus import crewai_event_bus
+    from crewai.events.types.flow_events import FlowPausedEvent
+    from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
+
+    persistence = SQLiteFlowPersistence(str(tmp_path / "flows.db"))
+
+    class State(BaseModel):
+        id: str = "resume-test-1"
+
+    class AsyncProvider:
+        def request_feedback(self, context: PendingFeedbackContext, flow: Flow) -> str:
+            raise HumanFeedbackPending(context=context)
+
+    class ReviewFlow(Flow[State]):
+        @start()
+        @human_feedback(message="Review:", provider=AsyncProvider())
+        def draft(self) -> str:
+            return "draft"
+
+        @listen(draft)
+        def finish(self, result) -> str:
+            return f"final: {result.feedback}"
+
+    paused: dict[str, str] = {}
+
+    @crewai_event_bus.on(FlowPausedEvent)
+    def _capture(source, event) -> None:
+        paused["flow_id"] = event.flow_id
+
+    with contextlib.suppress(BaseException):
+        ReviewFlow(persistence=persistence).kickoff()
+
+    assert ("ReviewFlow", "user") in pauses
+    assert starts == [("ReviewFlow", False)]
+
+    flow = ReviewFlow.from_pending(paused["flow_id"], persistence)
+    flow.resume("looks good")
+
+    assert ("ReviewFlow", True) in starts
+    assert ("ReviewFlow", "completed") in [(n, o) for n, _d, o in durations]
+
+
+def test_a_user_flow_that_suppresses_console_events_still_reports(
+    durations: list[tuple[str, float, str]],
+) -> None:
+    """``suppress_flow_events`` asks for console quiet, not for no telemetry."""
+
+    class QuietFlow(Flow):
+        suppress_flow_events: bool = True
+
+        @start()
+        def go(self) -> str:
+            return "ok"
+
+    QuietFlow().kickoff()
+
+    assert [(n, o) for n, _d, o in durations] == [("QuietFlow", "completed")]
+
+
+def test_a_declarative_flow_is_not_treated_as_internal(
+    flow_spans: list[tuple[str, str]],
+) -> None:
+    """``Flow.from_declaration()`` yields a ``Flow``, defined inside crewai.
+
+    Deciding origin from the defining module would report a caller's
+    declarative flow as one of CrewAI's own.
+    """
+    flow = Flow.from_declaration(contents={"name": "MyDeclarativeFlow"})
+
+    assert getattr(type(flow), "is_crewai_internal", False) is False
+
+
+def test_a_failed_conversation_session_is_not_reported_completed(
+    features: list[str], durations: list[tuple[str, float, str]]
+) -> None:
+    """A conversational session closes with FlowFinishedEvent either way.
+
+    Reading that event at face value counted a failed session as a success,
+    alongside the turn-failure signal.
+    """
+
+    class FailingChat(Flow):
+        conversational = True
+
+        @start()
+        def begin(self) -> str:
+            raise RuntimeError("turn exploded")
+
+    chat = FailingChat()
+    with pytest.raises(RuntimeError, match="turn exploded"):
+        chat.handle_turn("hello")
+    chat.finalize_session_traces()
+
+    assert "flow:conversation_turn_failed" in features
+    assert all(outcome != "completed" for _n, _d, outcome in durations)
+
+
+def test_infrastructure_flows_do_not_pollute_outcome_signals(
+    features: list[str], durations: list[tuple[str, float, str]]
+) -> None:
+    """CrewAI's own flows must not be counted as user flow outcomes.
+
+    The agent executor, memory encoding and memory recall are all Flows and run
+    far more often than anything a user wrote. Counting their outcomes in the
+    same feature would make ``flow:completed`` mostly bookkeeping. Their outcome
+    is still recorded on the Flow Completed span, which carries ``origin``.
+    """
+    from crewai import Agent, Crew, Task
+    from crewai.llms.base_llm import BaseLLM
+
+    class StubLLM(BaseLLM):
+        def __init__(self) -> None:
+            super().__init__(model="stub-model")
+
+        def call(self, messages, **kwargs) -> str:
+            return "Final Answer: done"
+
+        def supports_function_calling(self) -> bool:
+            return False
+
+        def supports_stop_words(self) -> bool:
+            return False
+
+        def get_context_window_size(self) -> int:
+            return 8192
+
+    agent = Agent(role="R", goal="G", backstory="B", llm=StubLLM())
+    task = Task(description="Do it", expected_output="A result", agent=agent)
+    Crew(agents=[agent], tasks=[task]).kickoff()
+
+    # Internal outcomes are still recorded - on the span, tagged internal -
+    # they simply do not masquerade as a user's flow finishing.
+    assert ("AgentExecutor", "completed") in [
+        (name, outcome) for name, _duration, outcome in durations
+    ]
+    assert "flow:completed" not in features
+
+
+def test_a_checkpoint_restore_is_not_counted_as_a_resume(
+    starts: list[tuple[str, bool]],
+) -> None:
+    """Only a run restored from a human pause is marked resumed.
+
+    ``_is_execution_resuming`` is also set by checkpoint restores that never
+    paused for anyone. Counting those would push resumes above pauses and make
+    the abandonment rate meaningless.
+    """
+    from crewai.events.event_bus import crewai_event_bus
+    from crewai.events.types.flow_events import FlowStartedEvent
+
+    class RestoredFlow(Flow):
+        @start()
+        def go(self) -> str:
+            return "ok"
+
+    flow = RestoredFlow()
+    flow._is_execution_resuming = True
+    assert flow._pending_feedback_context is None
+
+    crewai_event_bus.emit(flow, FlowStartedEvent(flow_name="RestoredFlow"))
+    wait_for_event_handlers()
+
+    assert starts == [("RestoredFlow", False)]

lib/crewai/tests/telemetry/test_telemetry.py

@@ -306,3 +306,68 @@ def test_event_listener_tracks_hook_dispatched_events():
         interception_point="pre_tool_call",
         outcome="aborted",
     )
+
+
+def _emit(method: str, *args, **kwargs):
+    """Run one telemetry span method against a mocked tracer.
+
+    The singleton is reset first: it caches the provider built on the very
+    first construction, so without this only the earliest caller in a session
+    would see the mocked tracer.
+    """
+    tracer = Mock()
+    span = Mock()
+    tracer.start_span.return_value = span
+    Telemetry._instance = None
+
+    with (
+        patch.dict(
+            os.environ,
+            {
+                "CREWAI_DISABLE_TELEMETRY": "false",
+                "CREWAI_DISABLE_TRACKING": "false",
+                "OTEL_SDK_DISABLED": "false",
+            },
+        ),
+        patch(
+            "crewai.telemetry.telemetry.TracerProvider",
+            return_value=Mock(get_tracer=Mock(return_value=tracer)),
+        ),
+        patch("crewai.telemetry.telemetry.version", return_value="9.9.9"),
+    ):
+        getattr(Telemetry(), method)(*args, **kwargs)
+    Telemetry._instance = None
+    return tracer, span
+
+
+@pytest.mark.parametrize(("resumed", "expected"), [(True, "true"), (False, "false")])
+def test_resumed_is_recorded_as_a_string(resumed: bool, expected: str) -> None:
+    """A boolean is encoded as the presence of a key, not as a value.
+
+    ``false`` arrives as the key simply being absent, which is invisible in the
+    schema and easy to extract wrongly - crew_memory reads 1 for 99.8% of crews
+    for exactly that reason. A string leaves nothing to infer.
+    """
+    _tracer, span = _emit(
+        "flow_execution_span", "ResearchFlow", ["start"], "user", resumed
+    )
+
+    span.set_attribute.assert_any_call("resumed", expected)
+    for call in span.set_attribute.call_args_list:
+        assert call.args[1] is not True and call.args[1] is not False
+
+
+def test_flow_completed_records_duration_outcome_and_origin() -> None:
+    _tracer, span = _emit("flow_completed_span", "ResearchFlow", 12.5, "failed", "user")
+
+    span.set_attribute.assert_any_call("flow_name", "ResearchFlow")
+    span.set_attribute.assert_any_call("duration_ms", 12.5)
+    span.set_attribute.assert_any_call("outcome", "failed")
+    span.set_attribute.assert_any_call("origin", "user")
+
+
+def test_paused_and_method_failed_record_flow_and_origin() -> None:
+    for method in ("flow_paused_span", "flow_method_failed_span"):
+        _tracer, span = _emit(method, "ResearchFlow", "internal")
+        span.set_attribute.assert_any_call("flow_name", "ResearchFlow")
+        span.set_attribute.assert_any_call("origin", "internal")

lib/crewai/tests/utilities/test_events.py

@@ -505,7 +505,9 @@ def test_flow_emits_start_event(reset_event_listener_singleton):
         flow.kickoff()
 
     assert event_received.wait(timeout=5), "Timeout waiting for flow started event"
-    mock_telemetry.flow_execution_span.assert_called_once_with("TestFlow", ["begin"])
+    mock_telemetry.flow_execution_span.assert_called_once_with(
+        "TestFlow", ["begin"], "user", False
+    )
     assert len(received_events) == 1
     assert received_events[0].flow_name == "TestFlow"
     assert received_events[0].type == "flow_started"