diff --git a/docs/edge/ar/telemetry.mdx b/docs/edge/ar/telemetry.mdx
index 21d22b997..e406afae1 100644
--- a/docs/edge/ar/telemetry.mdx
+++ b/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'. لا بيانات شخصية. |
diff --git a/docs/edge/en/telemetry.mdx b/docs/edge/en/telemetry.mdx
index 074930ade..c7dabddfd 100644
--- a/docs/edge/en/telemetry.mdx
+++ b/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. |
diff --git a/docs/edge/ko/telemetry.mdx b/docs/edge/ko/telemetry.mdx
index 1a61d308e..ee9ffa86c 100644
--- a/docs/edge/ko/telemetry.mdx
+++ b/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'. 개인 정보 없음. |
diff --git a/docs/edge/pt-BR/telemetry.mdx b/docs/edge/pt-BR/telemetry.mdx
index 5b7c1ea69..fffddd354 100644
--- a/docs/edge/pt-BR/telemetry.mdx
+++ b/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.                       |
diff --git a/lib/crewai/src/crewai/events/event_listener.py b/lib/crewai/src/crewai/events/event_listener.py
index d4b6a2d3f..8b0d789b6 100644
--- a/lib/crewai/src/crewai/events/event_listener.py
+++ b/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,
diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py
index 54fea7ca0..88794c654 100644
--- a/lib/crewai/src/crewai/experimental/agent_executor.py
+++ b/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
diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py
index 46278f4fe..30d5fbe19 100644
--- a/lib/crewai/src/crewai/flow/runtime/__init__.py
+++ b/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)
diff --git a/lib/crewai/src/crewai/memory/encoding_flow.py b/lib/crewai/src/crewai/memory/encoding_flow.py
index 968b439bf..1f197191e 100644
--- a/lib/crewai/src/crewai/memory/encoding_flow.py
+++ b/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
diff --git a/lib/crewai/src/crewai/memory/recall_flow.py b/lib/crewai/src/crewai/memory/recall_flow.py
index 562768c18..8b2d64651 100644
--- a/lib/crewai/src/crewai/memory/recall_flow.py
+++ b/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
diff --git a/lib/crewai/src/crewai/telemetry/telemetry.py b/lib/crewai/src/crewai/telemetry/telemetry.py
index ea2052acb..3b7a2682e 100644
--- a/lib/crewai/src/crewai/telemetry/telemetry.py
+++ b/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)
diff --git a/lib/crewai/tests/telemetry/test_flow_telemetry.py b/lib/crewai/tests/telemetry/test_flow_telemetry.py
new file mode 100644
index 000000000..df32834bf
--- /dev/null
+++ b/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)]
diff --git a/lib/crewai/tests/telemetry/test_telemetry.py b/lib/crewai/tests/telemetry/test_telemetry.py
index acc44bf9e..5c30bd4ff 100644
--- a/lib/crewai/tests/telemetry/test_telemetry.py
+++ b/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")
diff --git a/lib/crewai/tests/utilities/test_events.py b/lib/crewai/tests/utilities/test_events.py
index 6f2926b22..11a40679c 100644
--- a/lib/crewai/tests/utilities/test_events.py
+++ b/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"