From 8b646620bebbcb2ce625bb8e9a9f4e10be5e2510 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Moura?= Date: Wed, 12 Aug 2026 12:29:19 -0300 Subject: [PATCH] fix(events): stop a failed turn from marking the next one failed (#6965) * fix(telemetry): stop a failed turn from marking the next one failed A conversational session that opts out of deferred finalization ends each turn with its own FlowFailedEvent, emitted inside kickoff() before handle_turn() emits ConversationTurnFailedEvent. The flag was therefore set after the run that owned it had already cleared it, survived on the instance, and reported the next healthy turn as failed. Gate the flag on the run still having its start stamp: a deferring session keeps it (no per-turn terminal event), so it still reports a failed turn at session end. Also aligns the Flow Lifecycle Signals privacy row with the rest of the telemetry table, which qualifies every user-authored field it records with "should not include personal info", and fixes a telemetry test that built InputResponse with an unsupported `value` keyword - ask() swallowed the TypeError, so the test asserted the signals while exercising the provider-error path. Co-Authored-By: Claude Opus 5 (1M context) * test(telemetry): cover the streamed turn emitter of the failure flag stream_turn() is the second emitter of ConversationTurnFailedEvent and leaks the same flag as handle_turn(). Both regression tests fail on 77c68bd with ['failed', 'failed']. Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Opus 5 (1M context) --- docs/edge/ar/telemetry.mdx | 2 +- docs/edge/en/telemetry.mdx | 2 +- docs/edge/ko/telemetry.mdx | 2 +- docs/edge/pt-BR/telemetry.mdx | 2 +- .../src/crewai/events/event_listener.py | 29 +++- lib/crewai/src/crewai/telemetry/telemetry.py | 20 ++- .../tests/telemetry/test_flow_telemetry.py | 158 +++++++++++++++++- lib/crewai/tests/telemetry/test_telemetry.py | 11 ++ lib/crewai/tests/utilities/test_events.py | 2 +- 9 files changed, 209 insertions(+), 19 deletions(-) diff --git a/docs/edge/ar/telemetry.mdx b/docs/edge/ar/telemetry.mdx index e406afae13..398d8cc880 100644 --- a/docs/edge/ar/telemetry.mdx +++ b/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'. لا بيانات شخصية. | diff --git a/docs/edge/en/telemetry.mdx b/docs/edge/en/telemetry.mdx index c7dabddfd1..d3e94e3901 100644 --- a/docs/edge/en/telemetry.mdx +++ b/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: 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. | +| 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 (should not include personal info), 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 ee9ffa86c9..cf05bf31a7 100644 --- a/docs/edge/ko/telemetry.mdx +++ b/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'. 개인 정보 없음. | diff --git a/docs/edge/pt-BR/telemetry.mdx b/docs/edge/pt-BR/telemetry.mdx index fffddd354f..d38588b2c9 100644 --- a/docs/edge/pt-BR/telemetry.mdx +++ b/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: 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. | +| 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 (não deve incluir info pessoal), 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 8b0d789b6a..4182c5f049 100644 --- a/lib/crewai/src/crewai/events/event_listener.py +++ b/lib/crewai/src/crewai/events/event_listener.py @@ -308,6 +308,17 @@ class EventListener(BaseEventListener): def on_flow_created(_: Any, event: FlowCreatedEvent) -> None: self._telemetry.flow_creation_span(event.flow_name) + def _is_conversational(source: Any) -> bool: + """Whether this run is a turn of a conversational session. + + Reads the flow's own accessor, which covers both the + ``conversational = True`` class attribute and a conversational + definition. Each turn is its own kickoff while the session reports + a single completion, so these spans run many-to-one. + """ + checker = getattr(source, "_is_conversational_enabled", None) + return bool(checker()) if callable(checker) else False + def _flow_origin(source: Any) -> str: """Separate flows CrewAI runs itself from the ones a caller wrote. @@ -335,10 +346,8 @@ 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. + # Reset point for the flag a deferred session accumulates across + # turns, so a later session on this instance starts clean. source._telemetry_turn_failed = False started_at = getattr(source, "_telemetry_started_at", None) if started_at is None: @@ -349,6 +358,7 @@ class EventListener(BaseEventListener): (time.monotonic() - started_at) * 1000, outcome, _flow_origin(source), + _is_conversational(source), ) @crewai_event_bus.on(FlowStartedEvent) @@ -363,6 +373,7 @@ class EventListener(BaseEventListener): list(source._methods.keys()), _flow_origin(source), resumed, + _is_conversational(source), ) source._telemetry_started_at = time.monotonic() if not getattr(source, "suppress_flow_events", False): @@ -406,9 +417,13 @@ class EventListener(BaseEventListener): 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 + # A deferred session closes with FlowFinishedEvent whatever happened, + # so record the failure for on_flow_finished to read. Without + # deferral the run already emitted FlowFailedEvent before + # handle_turn emits this one - it cleared the stamp, and flagging + # now would mark the next turn on this instance failed. + if getattr(source, "_telemetry_started_at", None) is not None: + source._telemetry_turn_failed = True @crewai_event_bus.on(FlowInputRequestedEvent) def on_flow_input_requested(_: Any, event: FlowInputRequestedEvent) -> None: diff --git a/lib/crewai/src/crewai/telemetry/telemetry.py b/lib/crewai/src/crewai/telemetry/telemetry.py index 3b7a2682e4..19fedb346c 100644 --- a/lib/crewai/src/crewai/telemetry/telemetry.py +++ b/lib/crewai/src/crewai/telemetry/telemetry.py @@ -997,6 +997,7 @@ class Telemetry: node_names: list[str], origin: str = "user", resumed: bool = False, + conversational: bool = False, ) -> None: """Records the execution of a flow. @@ -1011,6 +1012,10 @@ class Telemetry: 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. + conversational: True for a turn of a conversational flow. Each turn + is its own kickoff but a session reports one completion, so + these spans run many-to-one and would otherwise drag any + completion rate computed across all flows. """ def _operation() -> None: @@ -1030,12 +1035,20 @@ class Telemetry: # 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") + self._add_attribute( + span, "conversational", "true" if conversational 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" + self, + flow_name: str, + duration_ms: float, + outcome: str, + origin: str = "user", + conversational: bool = False, ) -> None: """Records how long a flow ran and how it ended. @@ -1056,6 +1069,8 @@ class Telemetry: outcome: Either ``"completed"`` or ``"failed"``. origin: ``"internal"`` for flows CrewAI itself runs (the agent executor), ``"user"`` for flows the caller authored. + conversational: True when this closes a conversational session. One + of these answers many ``Flow Execution`` spans, one per turn. """ def _operation() -> None: @@ -1066,6 +1081,9 @@ class Telemetry: self._add_attribute(span, "duration_ms", duration_ms) self._add_attribute(span, "outcome", outcome) self._add_attribute(span, "origin", origin) + self._add_attribute( + span, "conversational", "true" if conversational else "false" + ) 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 index df32834bf6..81fbd03a17 100644 --- a/lib/crewai/tests/telemetry/test_flow_telemetry.py +++ b/lib/crewai/tests/telemetry/test_flow_telemetry.py @@ -17,6 +17,7 @@ import time import pytest +from crewai.experimental.conversational import ConversationConfig from crewai.flow.async_feedback import HumanFeedbackPending, PendingFeedbackContext from crewai.flow.flow import Flow, listen, start from crewai.flow.human_feedback import human_feedback @@ -55,7 +56,7 @@ 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", resumed=False: recorded.append( + lambda flow_name, node_names, origin="user", resumed=False, conversational=False: recorded.append( (flow_name, origin) ), ) @@ -73,13 +74,33 @@ def starts(monkeypatch: pytest.MonkeyPatch) -> 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( + lambda flow_name, node_names, origin="user", resumed=False, conversational=False: recorded.append( (flow_name, resumed) ), ) return recorded +@pytest.fixture +def conversational_marks(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, bool]]: + """Record (flow_name, conversational) 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, + conversational=False: recorded.append((flow_name, conversational)), + ) + return recorded + + @pytest.fixture def pauses(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str]]: """Record (flow_name, origin) for every Flow Paused span.""" @@ -123,7 +144,7 @@ def durations(monkeypatch: pytest.MonkeyPatch) -> 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( + lambda flow_name, duration_ms, outcome, origin="user", conversational=False: recorded.append( (flow_name, duration_ms, outcome) ), ) @@ -201,7 +222,7 @@ 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", resumed=False: started.append( + lambda flow_name, node_names, origin="user", resumed=False, conversational=False: started.append( flow_name ), ) @@ -220,14 +241,17 @@ def test_a_failed_flow_is_still_counted_as_an_execution( 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") + return InputResponse(text="typed answer") class AskFlow(Flow): @start() def go(self) -> str: return self.ask("What topic?") - AskFlow(input_provider=StubProvider()).kickoff() + # ask() swallows provider errors and returns None, so the answer is + # asserted too: a provider that raises would otherwise still emit both + # signals and pass this test. + assert AskFlow(input_provider=StubProvider()).kickoff() == "typed answer" emitted = features assert "flow:input_requested" in emitted @@ -549,6 +573,94 @@ def test_a_failed_conversation_session_is_not_reported_completed( assert all(outcome != "completed" for _n, _d, outcome in durations) +def test_a_deferred_session_still_reports_a_failed_turn( + durations: list[tuple[str, float, str]], +) -> None: + """A deferring session has no per-turn terminal event to carry the failure. + + Its only outcome span is the one ``finalize_session_traces()`` triggers, so + the turn-failure flag is what makes that span say ``failed``. Deferral is + the default for a conversational flow, so this is the common path. + """ + + class DeferringChat(Flow): + conversational = True + + @start() + def begin(self) -> str: + raise RuntimeError("turn exploded") + + chat = DeferringChat() + with pytest.raises(RuntimeError, match="turn exploded"): + chat.handle_turn("hello") + chat.finalize_session_traces() + # finalize_session_traces() emits without awaiting its handlers. + wait_for_event_handlers() + + assert [outcome for _n, _d, outcome in durations] == ["failed"] + + +def test_a_failed_turn_does_not_mark_the_next_turn_failed( + durations: list[tuple[str, float, str]], +) -> None: + """A session that opts out of deferral ends each turn with its own event. + + That terminal event fires inside ``kickoff()``, before ``handle_turn()`` + emits the turn-failure event, so the flag was set after the run that owned + it had already cleared it - and the next healthy turn read it as failed. + """ + + turns: list[str] = [] + + @ConversationConfig(defer_trace_finalization=False) + class FlakyChat(Flow): + conversational = True + + @start() + def begin(self) -> str: + turns.append("turn") + if len(turns) == 1: + raise RuntimeError("turn exploded") + return "second turn is fine" + + chat = FlakyChat() + with pytest.raises(RuntimeError, match="turn exploded"): + chat.handle_turn("hello") + chat.handle_turn("again") + + assert [outcome for _n, _d, outcome in durations] == ["failed", "completed"] + + +def test_a_failed_streamed_turn_does_not_mark_the_next_turn_failed( + durations: list[tuple[str, float, str]], +) -> None: + """``stream_turn`` is the other emitter of the turn-failure event. + + It emits from its own ``except`` block, after ``kickoff()`` has closed the + run out, so it leaks the same flag as the non-streamed path. + """ + + turns: list[str] = [] + + @ConversationConfig(defer_trace_finalization=False) + class FlakyStreamingChat(Flow): + conversational = True + + @start() + def begin(self) -> str: + turns.append("turn") + if len(turns) == 1: + raise RuntimeError("turn exploded") + return "second turn is fine" + + chat = FlakyStreamingChat() + with pytest.raises(RuntimeError, match="turn exploded"): + list(chat.stream_turn("hello").events) + list(chat.stream_turn("again").events) + + assert [outcome for _n, _d, outcome in durations] == ["failed", "completed"] + + def test_infrastructure_flows_do_not_pollute_outcome_signals( features: list[str], durations: list[tuple[str, float, str]] ) -> None: @@ -615,3 +727,37 @@ def test_a_checkpoint_restore_is_not_counted_as_a_resume( wait_for_event_handlers() assert starts == [("RestoredFlow", False)] + + +def test_a_conversational_turn_is_marked( + conversational_marks: list[tuple[str, bool]], +) -> None: + """Each turn is its own kickoff, but a session reports one completion. + + Without the marker those spans run many-to-one against Flow Completed and + silently drag any completion rate computed across all flows. + """ + + class Chatty(Flow): + conversational = True + + @start() + def begin(self) -> str: + return "hi" + + Chatty().handle_turn("hello") + + assert ("Chatty", True) in conversational_marks + + +def test_an_ordinary_flow_is_not_marked_conversational( + conversational_marks: list[tuple[str, bool]], +) -> None: + class PlainFlow(Flow): + @start() + def go(self) -> str: + return "ok" + + PlainFlow().kickoff() + + assert ("PlainFlow", False) in conversational_marks diff --git a/lib/crewai/tests/telemetry/test_telemetry.py b/lib/crewai/tests/telemetry/test_telemetry.py index 5c30bd4ffd..70b0fd0e8d 100644 --- a/lib/crewai/tests/telemetry/test_telemetry.py +++ b/lib/crewai/tests/telemetry/test_telemetry.py @@ -364,6 +364,17 @@ def test_flow_completed_records_duration_outcome_and_origin() -> None: 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") + span.set_attribute.assert_any_call("conversational", "false") + + +@pytest.mark.parametrize(("flag", "expected"), [(True, "true"), (False, "false")]) +def test_conversational_is_recorded_as_a_string(flag: bool, expected: str) -> None: + """Same reason as resumed: a bool arrives as key presence, not a value.""" + _tracer, span = _emit( + "flow_execution_span", "ResearchFlow", ["start"], "user", False, flag + ) + + span.set_attribute.assert_any_call("conversational", expected) def test_paused_and_method_failed_record_flow_and_origin() -> None: diff --git a/lib/crewai/tests/utilities/test_events.py b/lib/crewai/tests/utilities/test_events.py index 11a40679ca..f4bf43100c 100644 --- a/lib/crewai/tests/utilities/test_events.py +++ b/lib/crewai/tests/utilities/test_events.py @@ -506,7 +506,7 @@ def test_flow_emits_start_event(reset_event_listener_singleton): assert event_received.wait(timeout=5), "Timeout waiting for flow started event" mock_telemetry.flow_execution_span.assert_called_once_with( - "TestFlow", ["begin"], "user", False + "TestFlow", ["begin"], "user", False, False ) assert len(received_events) == 1 assert received_events[0].flow_name == "TestFlow"