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

docs/edge/ar/telemetry.mdx

@@ -61,7 +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 داخليًا أم من كتابتك. يُسجَّل اسم التدفق، كما هو الحال بالفعل عند إنشاء التدفق وتنفيذه. لا تُسجَّل أبدًا أسماء الدوال أو رسائل الأخطاء أو حالة التدفق. لا بيانات شخصية. |
+| نعم       | إشارات دورة حياة التدفق | تشمل: بدء التدفق، وما إذا اكتمل أو فشل، وما إذا فشلت إحدى دواله، وما إذا توقّف مؤقتًا لطلب إدخال أو ملاحظات بشرية، وما إذا كان البدء استئنافًا لتشغيل سابق، وما إذا فشلت دورة محادثة، ومدة تشغيل التدفق، وما إذا كان التدفق من التدفقات التي يشغّلها CrewAI داخليًا أم من كتابتك. يُسجَّل اسم التدفق، كما هو الحال بالفعل عند إنشاء التدفق وتنفيذه. لا تُسجَّل أبدًا أسماء الدوال أو رسائل الأخطاء أو حالة التدفق. لا بيانات شخصية. |
 | لا        | بيانات الوكيل الموسّعة | تشمل: وصف الهدف، نص الخلفية، معرّف ملف موجهات i18n. يجب على المستخدمين التأكد من عدم تضمين معلومات شخصية في حقول النص. |
 | لا        | معلومات المهمة التفصيلية | تشمل: وصف المهمة، وصف المخرجات المتوقعة، مراجع السياق. يجب على المستخدمين التأكد من عدم تضمين معلومات شخصية في هذه الحقول. |
 | لا        | معلومات البيئة | تشمل: المنصة، الإصدار، النظام، الإصدار، وعدد وحدات المعالجة المركزية. مثال: 'Windows 10'، 'x86_64'. لا بيانات شخصية. |

docs/edge/en/telemetry.mdx

@@ -61,7 +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: whether a flow completed or failed, whether one of its methods failed, whether it paused for human input or feedback, whether it was resumed afterwards, 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. |
+| 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,7 +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 상태는 절대 기록하지 않습니다. 개인 정보 없음. |
+| 예      | Flow 라이프사이클 신호 | 포함 항목: flow의 시작, 완료 또는 실패 여부, 해당 메서드의 실패 여부, 사람의 입력이나 피드백을 위해 일시 중지되었는지 여부, 해당 시작이 재개된 실행인지 여부, 대화 턴의 실패 여부, flow 실행 시간, 그리고 해당 flow가 CrewAI가 내부적으로 실행하는 것인지 사용자가 작성한 것인지 여부. flow 이름은 기록되며, 이는 flow 생성 및 실행에서 이미 그러합니다. 메서드 이름, 오류 메시지, flow 상태는 절대 기록하지 않습니다. 개인 정보 없음. |
 | 아니오  | 에이전트 확장 데이터                      | 목표 설명, 배경 이야기 텍스트, i18n 프롬프트 파일 식별자가 포함됩니다. 사용자들은 텍스트 필드에 개인 정보가 포함되지 않도록 해야 합니다. |
 | 아니오  | 상세 작업 정보                            | 작업 설명, 예상 출력 설명, 컨텍스트 참조가 포함됩니다. 사용자들은 이러한 필드에 개인 정보가 포함되지 않도록 해야 합니다. |
 | 아니오  | 환경 정보                                 | 플랫폼, 릴리즈, 시스템, 버전, CPU 개수가 포함됩니다. 예: 'Windows 10', 'x86_64'. 개인 정보 없음. |

docs/edge/pt-BR/telemetry.mdx

@@ -61,7 +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: se um flow foi concluído ou falhou, se um de seus métodos falhou, se ele pausou para entrada ou feedback humano, se foi retomado em seguida, 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. |
+| 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

@@ -335,6 +335,11 @@ class EventListener(BaseEventListener):
             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
@@ -348,17 +353,18 @@ class EventListener(BaseEventListener):
 
         @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()), _flow_origin(source)
+                event.flow_name,
+                list(source._methods.keys()),
+                _flow_origin(source),
+                resumed,
             )
             source._telemetry_started_at = time.monotonic()
-            if getattr(source, "_pending_feedback_context", None) is not None:
-                # No resume event exists, so a run restored from a pause is only
-                # visible here. Keyed off the pending-feedback context rather
-                # than _is_execution_resuming, which is also set by checkpoint
-                # restores that never paused for a human - counting those would
-                # inflate resumes past pauses and break the abandonment rate.
-                self._telemetry.feature_usage_span("flow:resumed")
             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))
@@ -370,9 +376,6 @@ class EventListener(BaseEventListener):
                 if getattr(source, "_telemetry_turn_failed", False)
                 else "completed"
             )
-            source._telemetry_turn_failed = False
-            if _flow_origin(source) == "user":
-                self._telemetry.feature_usage_span(f"flow:{outcome}")
             _report_flow_duration(source, event.flow_name, outcome)
 
             if not getattr(source, "suppress_flow_events", False):
@@ -383,8 +386,6 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(FlowFailedEvent)
         def on_flow_failed(source: Any, event: FlowFailedEvent) -> None:
-            if _flow_origin(source) == "user":
-                self._telemetry.feature_usage_span("flow:failed")
             _report_flow_duration(source, event.flow_name, "failed")
 
             if not getattr(source, "suppress_flow_events", False):
@@ -445,8 +446,9 @@ class EventListener(BaseEventListener):
         ) -> None:
             # The method name is not recorded: it is user-authored and would put
             # arbitrary strings in telemetry.
-            if _flow_origin(source) == "user":
-                self._telemetry.feature_usage_span("flow:method_failed")
+            self._telemetry.flow_method_failed_span(
+                event.flow_name, _flow_origin(source)
+            )
 
             self.formatter.handle_method_status(
                 event.method_name,
@@ -465,8 +467,8 @@ class EventListener(BaseEventListener):
             )
 
         @crewai_event_bus.on(FlowPausedEvent)
-        def on_flow_paused(_: Any, event: FlowPausedEvent) -> None:
-            self._telemetry.feature_usage_span("flow:paused")
+        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,

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

@@ -992,7 +992,11 @@ class Telemetry:
         self._safe_telemetry_operation(_operation)
 
     def flow_execution_span(
-        self, flow_name: str, node_names: list[str], origin: str = "user"
+        self,
+        flow_name: str,
+        node_names: list[str],
+        origin: str = "user",
+        resumed: bool = False,
     ) -> None:
         """Records the execution of a flow.
 
@@ -1003,6 +1007,10 @@ class Telemetry:
                 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:
@@ -1016,6 +1024,7 @@ 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)
+            self._add_attribute(span, "resumed", resumed)
             close_span(span)
 
         self._safe_telemetry_operation(_operation)
@@ -1056,6 +1065,51 @@ class Telemetry:
 
         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)
+
     def env_context_span(self, tool: str) -> None:
         """Records the coding tool environment context."""
 

lib/crewai/tests/telemetry/test_flow_telemetry.py

@@ -53,13 +53,63 @@ def flow_spans(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str]]:
     monkeypatch.setattr(
         listener_module.event_listener._telemetry,
         "flow_execution_span",
-        lambda flow_name, node_names, origin="user": recorded.append(
+        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."""
@@ -101,7 +151,11 @@ def features(monkeypatch: pytest.MonkeyPatch) -> list[str]:
     return recorded
 
 
-def test_completed_flow_reports_its_outcome(features: list[str]) -> None:
+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:
@@ -109,10 +163,12 @@ def test_completed_flow_reports_its_outcome(features: list[str]) -> None:
 
     OkFlow().kickoff()
 
-    assert "flow:completed" in features
+    assert [(n, o) for n, _d, o in durations] == [("OkFlow", "completed")]
 
 
-def test_failed_flow_reports_the_failure_and_the_method(features: list[str]) -> None:
+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:
@@ -121,10 +177,8 @@ def test_failed_flow_reports_the_failure_and_the_method(features: list[str]) ->
     with pytest.raises(RuntimeError, match="boom"):
         BoomFlow().kickoff()
 
-    emitted = features
-    assert "flow:failed" in emitted
-    assert "flow:method_failed" in emitted
-    assert "flow:completed" not in emitted
+    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(
@@ -145,7 +199,9 @@ def test_a_failed_flow_is_still_counted_as_an_execution(
     monkeypatch.setattr(
         listener_module.event_listener._telemetry,
         "flow_execution_span",
-        lambda flow_name, node_names, origin="user": started.append(flow_name),
+        lambda flow_name, node_names, origin="user", resumed=False: started.append(
+            flow_name
+        ),
     )
 
     class BoomFlow(Flow):
@@ -176,7 +232,9 @@ def test_requesting_input_reports_both_sides(features: list[str]) -> None:
     assert "flow:input_received" in emitted
 
 
-def test_paused_flow_reports_the_pause(features: list[str]) -> None:
+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:
@@ -198,9 +256,10 @@ def test_paused_flow_reports_the_pause(features: list[str]) -> None:
     with contextlib.suppress(BaseException):
         PausingFlow().kickoff()
 
-    emitted = features
-    assert "flow:hitl_paused" in emitted
-    assert "flow:paused" in emitted
+    # 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:
@@ -219,8 +278,16 @@ def test_failed_conversation_turn_is_reported(features: list[str]) -> None:
     assert "flow:conversation_turn_failed" in features
 
 
-def test_no_user_authored_strings_are_recorded(features: list[str]) -> None:
-    """Method names, flow names and error text must not reach telemetry."""
+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()
@@ -230,12 +297,15 @@ def test_no_user_authored_strings_are_recorded(features: list[str]) -> None:
     with pytest.raises(RuntimeError, match="secret error detail"):
         SecretNamedFlow().kickoff()
 
-    emitted = features
-    assert emitted
-    for feature in emitted:
-        assert "my_secret_method_name" not in feature
-        assert "secret error detail" not in feature
-        assert "SecretNamedFlow" not in feature
+    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(
@@ -367,7 +437,12 @@ def test_crewais_own_agent_executor_is_tagged_internal(flow_spans) -> None:
     assert origins.get("AgentExecutor") == "internal"
 
 
-def test_resumed_flow_is_reported(tmp_path, features: list[str]) -> None:
+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
@@ -407,18 +482,18 @@ def test_resumed_flow_is_reported(tmp_path, features: list[str]) -> None:
     with contextlib.suppress(BaseException):
         ReviewFlow(persistence=persistence).kickoff()
 
-    assert "flow:paused" in features
-    assert "flow:resumed" not in features
+    assert ("ReviewFlow", "user") in pauses
+    assert starts == [("ReviewFlow", False)]
 
     flow = ReviewFlow.from_pending(paused["flow_id"], persistence)
     flow.resume("looks good")
 
-    assert "flow:resumed" in features
-    assert "flow:completed" in features
+    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(
-    features: list[str],
+    durations: list[tuple[str, float, str]],
 ) -> None:
     """``suppress_flow_events`` asks for console quiet, not for no telemetry."""
 
@@ -431,7 +506,7 @@ def test_a_user_flow_that_suppresses_console_events_still_reports(
 
     QuietFlow().kickoff()
 
-    assert "flow:completed" in features
+    assert [(n, o) for n, _d, o in durations] == [("QuietFlow", "completed")]
 
 
 def test_a_declarative_flow_is_not_treated_as_internal(
@@ -469,7 +544,6 @@ def test_a_failed_conversation_session_is_not_reported_completed(
     chat.finalize_session_traces()
 
     assert "flow:conversation_turn_failed" in features
-    assert "flow:completed" not in features
     assert all(outcome != "completed" for _n, _d, outcome in durations)
 
 
@@ -506,10 +580,12 @@ def test_infrastructure_flows_do_not_pollute_outcome_signals(
     task = Task(description="Do it", expected_output="A result", agent=agent)
     Crew(agents=[agent], tasks=[task]).kickoff()
 
-    assert "flow:completed" not in features
+    # 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(