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

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

@@ -308,26 +308,24 @@ class EventListener(BaseEventListener):
         def on_flow_created(_: Any, event: FlowCreatedEvent) -> None:
             self._telemetry.flow_creation_span(event.flow_name)
 
-        def _is_infrastructure_flow(source: Any) -> bool:
-            """Flows CrewAI runs for its own bookkeeping.
-
-            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 counting their outcomes in the same
-            feature as user flows makes that feature meaningless. Their outcome
-            is still recorded on the Flow Completed span, which carries origin.
-            """
-            return bool(getattr(source, "suppress_flow_events", False))
-
         def _flow_origin(source: Any) -> str:
-            """Separate CrewAI's own flows from the caller's.
+            """Separate flows CrewAI runs itself from the ones a caller wrote.
 
-            The agent executor is itself a Flow and runs once per agent
-            execution, so it dominates flow counts and would otherwise be
-            indistinguishable from flows a user 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 type(source).__module__.startswith("crewai.") else "user"
+                "internal"
+                if getattr(type(source), "is_crewai_internal", False)
+                else "user"
             )
 
         def _report_flow_duration(source: Any, flow_name: str, outcome: str) -> None:
@@ -367,9 +365,15 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(FlowFinishedEvent)
         def on_flow_finished(source: Any, event: FlowFinishedEvent) -> None:
-            if not _is_infrastructure_flow(source):
-                self._telemetry.feature_usage_span("flow:completed")
-            _report_flow_duration(source, event.flow_name, "completed")
+            outcome = (
+                "failed"
+                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):
                 self.formatter.handle_flow_status(
@@ -379,7 +383,7 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(FlowFailedEvent)
         def on_flow_failed(source: Any, event: FlowFailedEvent) -> None:
-            if not _is_infrastructure_flow(source):
+            if _flow_origin(source) == "user":
                 self._telemetry.feature_usage_span("flow:failed")
             _report_flow_duration(source, event.flow_name, "failed")
 
@@ -398,9 +402,12 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(ConversationTurnFailedEvent)
         def on_conversation_turn_failed(
-            _: Any, event: ConversationTurnFailedEvent
+            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:
@@ -438,7 +445,7 @@ class EventListener(BaseEventListener):
         ) -> None:
             # The method name is not recorded: it is user-authored and would put
             # arbitrary strings in telemetry.
-            if not _is_infrastructure_flow(source):
+            if _flow_origin(source) == "user":
                 self._telemetry.feature_usage_span("flow:method_failed")
 
             self.formatter.handle_method_status(

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,6 +717,18 @@ 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

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/tests/telemetry/test_flow_telemetry.py

@@ -417,6 +417,62 @@ def test_resumed_flow_is_reported(tmp_path, features: list[str]) -> None:
     assert "flow:completed" in features
 
 
+def test_a_user_flow_that_suppresses_console_events_still_reports(
+    features: list[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 "flow:completed" in features
+
+
+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 "flow:completed" not 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: