From 1ca4762a80e8a2de365413b73e4ce8fee8f2c42b Mon Sep 17 00:00:00 2001 From: Joao Moura Date: Tue, 11 Aug 2026 16:06:23 -0700 Subject: [PATCH] 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) Claude-Session: https://claude.ai/code/session_01ASfWmW3RGy4qAQm6s8U9jH --- .../src/crewai/events/event_listener.py | 51 +++++++++-------- .../src/crewai/experimental/agent_executor.py | 3 +- .../src/crewai/flow/runtime/__init__.py | 12 ++++ lib/crewai/src/crewai/memory/encoding_flow.py | 4 +- lib/crewai/src/crewai/memory/recall_flow.py | 4 +- .../tests/telemetry/test_flow_telemetry.py | 56 +++++++++++++++++++ 6 files changed, 105 insertions(+), 25 deletions(-) diff --git a/lib/crewai/src/crewai/events/event_listener.py b/lib/crewai/src/crewai/events/event_listener.py index 1cb7b98dd7..985621fc02 100644 --- a/lib/crewai/src/crewai/events/event_listener.py +++ b/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( diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index 54fea7ca01..88794c654f 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 23bba18fb5..30d5fbe191 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/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 diff --git a/lib/crewai/src/crewai/memory/encoding_flow.py b/lib/crewai/src/crewai/memory/encoding_flow.py index 968b439bff..1f197191e9 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 562768c181..8b2d64651c 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/tests/telemetry/test_flow_telemetry.py b/lib/crewai/tests/telemetry/test_flow_telemetry.py index 8fc78d573b..09c5ee43bb 100644 --- a/lib/crewai/tests/telemetry/test_flow_telemetry.py +++ b/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: