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

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

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

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

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

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ASfWmW3RGy4qAQm6s8U9jH
This commit is contained in:
Joao Moura
2026-08-11 13:46:49 -07:00
parent 65a4b7cede
commit bb28ca5e3b
6 changed files with 234 additions and 0 deletions

View File

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

View File

@@ -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: whether a flow completed or failed, whether one of its methods failed, whether it paused for human input or feedback, and whether a conversation turn failed. Counts only - flow names, 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. |

View File

@@ -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 상태는 절대 기록하지 않습니다. 개인 정보 없음. |
| 아니오 | 에이전트 확장 데이터 | 목표 설명, 배경 이야기 텍스트, i18n 프롬프트 파일 식별자가 포함됩니다. 사용자들은 텍스트 필드에 개인 정보가 포함되지 않도록 해야 합니다. |
| 아니오 | 상세 작업 정보 | 작업 설명, 예상 출력 설명, 컨텍스트 참조가 포함됩니다. 사용자들은 이러한 필드에 개인 정보가 포함되지 않도록 해야 합니다. |
| 아니오 | 환경 정보 | 플랫폼, 릴리즈, 시스템, 버전, CPU 개수가 포함됩니다. 예: 'Windows 10', 'x86_64'. 개인 정보 없음. |

View File

@@ -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: se um flow foi concluído ou falhou, se um de seus métodos falhou, se ele pausou para entrada ou feedback humano e se um turno de conversa falhou. Apenas contagens - nomes de flows, 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. |

View File

@@ -42,9 +42,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,
@@ -315,6 +318,8 @@ class EventListener(BaseEventListener):
@crewai_event_bus.on(FlowFinishedEvent)
def on_flow_finished(source: Any, event: FlowFinishedEvent) -> None:
self._telemetry.feature_usage_span("flow:completed")
if not getattr(source, "suppress_flow_events", False):
self.formatter.handle_flow_status(
event.flow_name,
@@ -323,6 +328,8 @@ class EventListener(BaseEventListener):
@crewai_event_bus.on(FlowFailedEvent)
def on_flow_failed(source: Any, event: FlowFailedEvent) -> None:
self._telemetry.feature_usage_span("flow:failed")
if not getattr(source, "suppress_flow_events", False):
self.formatter.handle_flow_status(
event.flow_name,
@@ -336,6 +343,20 @@ class EventListener(BaseEventListener):
) -> None:
self._telemetry.feature_usage_span("flow:conversation_turn")
@crewai_event_bus.on(ConversationTurnFailedEvent)
def on_conversation_turn_failed(
_: Any, event: ConversationTurnFailedEvent
) -> None:
self._telemetry.feature_usage_span("flow:conversation_turn_failed")
@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
@@ -362,6 +383,10 @@ class EventListener(BaseEventListener):
def on_method_execution_failed(
_: Any, event: MethodExecutionFailedEvent
) -> None:
# The method name is not recorded: it is user-authored and would put
# arbitrary strings in telemetry.
self._telemetry.feature_usage_span("flow:method_failed")
self.formatter.handle_method_status(
event.method_name,
"failed",
@@ -371,6 +396,8 @@ 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",
@@ -378,6 +405,8 @@ class EventListener(BaseEventListener):
@crewai_event_bus.on(FlowPausedEvent)
def on_flow_paused(_: Any, event: FlowPausedEvent) -> None:
self._telemetry.feature_usage_span("flow:paused")
self.formatter.handle_flow_status(
event.flow_name,
event.flow_id,

View File

@@ -0,0 +1,201 @@
"""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 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
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 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(features: list[str]) -> None:
class OkFlow(Flow):
@start()
def go(self) -> str:
return "ok"
OkFlow().kickoff()
assert "flow:completed" in features
def test_failed_flow_reports_the_failure_and_the_method(features: list[str]) -> None:
class BoomFlow(Flow):
@start()
def go(self) -> str:
raise RuntimeError("boom")
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
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: 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]) -> 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()
emitted = features
assert "flow:hitl_paused" in emitted
assert "flow:paused" in emitted
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_user_authored_strings_are_recorded(features: list[str]) -> None:
"""Method names, flow names and error text must not reach telemetry."""
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()
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