Build flow state from FlowDefinition

Definition-driven flows previously always started with a bare dict
state.

.github/workflows/vulnerability-scan.yml

@@ -64,7 +64,6 @@ jobs:
             --ignore-vuln PYSEC-2025-197 \
             --ignore-vuln PYSEC-2025-210 \
             --ignore-vuln PYSEC-2026-139 \
-            --ignore-vuln GHSA-rrmf-rvhw-rf47 \
             --ignore-vuln PYSEC-2025-211 \
             --ignore-vuln PYSEC-2025-212 \
             --ignore-vuln PYSEC-2025-213 \
@@ -82,7 +81,6 @@ jobs:
         #   PYSEC-2025-183      - pyjwt 2.12.1: disputed weak-encryption claim; key length is application-chosen
         #   PYSEC-2025-189..197 - torch 2.11.0: memory-corruption/DoS in functions only reachable via untrusted models; no fix available
         #   PYSEC-2025-210, PYSEC-2026-139 - torch 2.11.0: profiler/deserialization issues; no fix available
-        #   GHSA-rrmf-rvhw-rf47 - torch 2.11.0 (CVE-2025-3000, alias of PYSEC-2025-194): memory corruption in torch.jit.script, CVSS 1.9, local-only; affected <=2.12.0, no fix available. pip-audit reports it under the GHSA id so the PYSEC ignore above does not catch it.
         #   PYSEC-2025-211..218 - transformers 5.5.4: deserialization/code injection via malicious model checkpoints; no fix available
         #   GHSA-f4j7-r4q5-qw2c - chromadb 1.1.1 (CVE-2026-45829): pre-auth RCE via /api/v2/tenants/{tenant}/databases/{db}/collections when trust_remote_code=true.
         #                         Advisory: vulnerable >=1.0.0,<=1.5.9, firstPatchedVersion=none. We only use chromadb.PersistentClient (lib/crewai/src/crewai/rag/chromadb/factory.py)

.pre-commit-config.yaml

@@ -47,7 +47,6 @@ repos:
           --ignore-vuln PYSEC-2025-197
           --ignore-vuln PYSEC-2025-210
           --ignore-vuln PYSEC-2026-139
-          --ignore-vuln GHSA-rrmf-rvhw-rf47
           --ignore-vuln PYSEC-2025-211
           --ignore-vuln PYSEC-2025-212
           --ignore-vuln PYSEC-2025-213

docs/ar/changelog.mdx

@@ -4,38 +4,6 @@ description: "تحديثات المنتج والتحسينات وإصلاحات
 icon: "clock"
 mode: "wide"
 ---
-<Update label="10 يونيو 2026">
-  ## v1.14.7rc1
-
-  [عرض الإصدار على GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.14.7rc1)
-
-  ## ما الذي تغير
-
-  ### الميزات
-  - إضافة `reset_runtime_state` لإطلاق حالة الحافلة المتراكمة
-  - التعامل مع دعم كل من الموجهات المخصصة
-  - فصل منطق المحادثة عن وقت التشغيل وإضافة `conversational_definition`
-
-  ### إصلاحات الأخطاء
-  - إصلاح نطاق حالة وقت التشغيل لكل تشغيل للحد من النمو وعزل التشغيلات المتزامنة
-  - إصلاح إعدادات القياس عن بُعد على `crewai-login`
-  - إصلاح احترام `suppress_flow_events` لفعاليات تنفيذ الأساليب
-
-  ### الوثائق
-  - تحديث صور OpenTelemetry
-  - تحديث الوثائق لتعكس الحالة الجديدة لجمع بيانات OpenTelemetry
-  - تحديث سجل التغييرات والإصدار لـ v1.14.7a4
-
-  ### إعادة الهيكلة
-  - تبسيط تقييم شرط التدفق ليكون بلا حالة لكل حدث
-  - تحسين دورة توجيه المحادثة مع تقليل مسار واحد
-
-  ## المساهمون
-
-  @greysonlalonde, @lorenzejay, @lucasgomide, @vinibrsl
-
-</Update>
-
 <Update label="9 يونيو 2026">
   ## v1.14.7a4
 

docs/ar/guides/advanced/customizing-prompts.mdx

@@ -161,18 +161,6 @@ crew = Crew(
 )
 ```
 
-<Note>
-يُحتفظ بـ `agent.i18n` للتوافق مع الإصدارات السابقة فقط، وقد تم إهماله. لتخصيص المطالبات أثناء التشغيل، مرّر `prompt_file` إلى `Crew`. وللوصول البرمجي المباشر إلى شرائح المطالبات، استخدم أداة i18n مباشرة:
-</Note>
-
-```python
-from crewai.utilities.i18n import get_i18n
-
-i18n = get_i18n("custom_prompts.json")
-format_slice = i18n.slice("format")
-tool_prompt = i18n.tools("ask_question")
-```
-
 #### الخيار 3: تعطيل مطالبات النظام لنماذج o1
 ```python
 agent = Agent(
@@ -220,8 +208,6 @@ agent = Agent(
 
 يدمج CrewAI بعد ذلك تخصيصاتك مع الإعدادات الافتراضية، فلا تحتاج لإعادة تعريف كل مطالبة. إليك الطريقة:
 
-بالنسبة للكود الذي يحتاج إلى قراءة شرائح المطالبات مباشرة، استخدم `crewai.utilities.i18n.get_i18n()` مع ملف المطالبات نفسه بدلًا من قراءة `agent.i18n`.
-
 ### مثال: تخصيص أساسي للمطالبات
 
 أنشئ ملف `custom_prompts.json` بالمطالبات التي تريد تعديلها. تأكد من إدراج جميع المطالبات عالية المستوى التي يجب أن يحتويها، وليس فقط تغييراتك:

docs/en/changelog.mdx

@@ -4,38 +4,6 @@ description: "Product updates, improvements, and bug fixes for CrewAI"
 icon: "clock"
 mode: "wide"
 ---
-<Update label="Jun 10, 2026">
-  ## v1.14.7rc1
-
-  [View release on GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.14.7rc1)
-
-  ## What's Changed
-
-  ### Features
-  - Add `reset_runtime_state` to release accumulated bus state
-  - Handle supporting both custom prompts
-  - Decouple conversation logic from runtime and add a `conversational_definition`
-
-  ### Bug Fixes
-  - Fix scope of runtime state per run to bound growth and isolate concurrent runs
-  - Fix telemetry setup on `crewai-login`
-  - Fix respect for `suppress_flow_events` for method-execution events
-
-  ### Documentation
-  - Update OpenTelemetry images
-  - Update documentation to reflect new state of OpenTelemetry collector
-  - Update changelog and version for v1.14.7a4
-
-  ### Refactoring
-  - Simplify flow condition evaluation to be stateless per event
-  - Improve conversation routing cycle with one less route
-
-  ## Contributors
-
-  @greysonlalonde, @lorenzejay, @lucasgomide, @vinibrsl
-
-</Update>
-
 <Update label="Jun 09, 2026">
   ## v1.14.7a4
 

docs/en/guides/advanced/customizing-prompts.mdx

@@ -161,18 +161,6 @@ crew = Crew(
 )
 ```
 
-<Note>
-`agent.i18n` is maintained only for backward compatibility and is deprecated. For runtime prompt customization, pass `prompt_file` to `Crew`. For programmatic access to prompt slices, use the i18n utility directly:
-</Note>
-
-```python
-from crewai.utilities.i18n import get_i18n
-
-i18n = get_i18n("custom_prompts.json")
-format_slice = i18n.slice("format")
-tool_prompt = i18n.tools("ask_question")
-```
-
 #### Option 3: Disable System Prompts for o1 Models
 ```python
 agent = Agent(
@@ -220,8 +208,6 @@ One straightforward approach is to create a JSON file for the prompts you want t
 
 CrewAI then merges your customizations with the defaults, so you don't have to redefine every prompt. Here's how:
 
-For code that needs to read prompt slices directly, use `crewai.utilities.i18n.get_i18n()` with the same prompt file instead of reading `agent.i18n`.
-
 ### Example: Basic Prompt Customization
 
 Create a `custom_prompts.json` file with the prompts you want to modify. Ensure you list all top-level prompts it should contain, not just your changes:

docs/ko/changelog.mdx

@@ -4,38 +4,6 @@ description: "CrewAI의 제품 업데이트, 개선 사항 및 버그 수정"
 icon: "clock"
 mode: "wide"
 ---
-<Update label="2026년 6월 10일">
-  ## v1.14.7rc1
-
-  [GitHub 릴리스 보기](https://github.com/crewAIInc/crewAI/releases/tag/1.14.7rc1)
-
-  ## 변경 사항
-
-  ### 기능
-  - 누적된 버스 상태를 해제하기 위해 `reset_runtime_state` 추가
-  - 사용자 정의 프롬프트를 모두 지원하도록 처리
-  - 대화 논리를 런타임과 분리하고 `conversational_definition` 추가
-
-  ### 버그 수정
-  - 실행당 런타임 상태의 범위를 수정하여 성장 제한 및 동시 실행 격리
-  - `crewai-login`에서 원격 측정 설정 수정
-  - 메서드 실행 이벤트에 대한 `suppress_flow_events` 존중 수정
-
-  ### 문서
-  - OpenTelemetry 이미지 업데이트
-  - OpenTelemetry 수집기의 새로운 상태를 반영하도록 문서 업데이트
-  - v1.14.7a4에 대한 변경 로그 및 버전 업데이트
-
-  ### 리팩토링
-  - 이벤트당 상태 비저장 방식으로 흐름 조건 평가 단순화
-  - 경로를 하나 줄여 대화 라우팅 사이클 개선
-
-  ## 기여자
-
-  @greysonlalonde, @lorenzejay, @lucasgomide, @vinibrsl
-
-</Update>
-
 <Update label="2026년 6월 9일">
   ## v1.14.7a4
 

docs/ko/guides/advanced/customizing-prompts.mdx

@@ -161,18 +161,6 @@ crew = Crew(
 )
 ```
 
-<Note>
-`agent.i18n`은 이전 버전과의 호환성을 위해서만 유지되며 사용이 중단될 예정입니다. 런타임 프롬프트 커스터마이징에는 `Crew`에 `prompt_file`을 전달하세요. 프롬프트 슬라이스를 코드에서 직접 읽어야 한다면 i18n 유틸리티를 직접 사용하세요:
-</Note>
-
-```python
-from crewai.utilities.i18n import get_i18n
-
-i18n = get_i18n("custom_prompts.json")
-format_slice = i18n.slice("format")
-tool_prompt = i18n.tools("ask_question")
-```
-
 #### 옵션 3: o1 모델에 대한 시스템 프롬프트 비활성화
 ```python
 agent = Agent(
@@ -220,8 +208,6 @@ agent = Agent(
 
 그러면 CrewAI가 기본값과 사용자가 지정한 내용을 병합하므로, 모든 프롬프트를 다시 정의할 필요가 없습니다. 방법은 다음과 같습니다:
 
-프롬프트 슬라이스를 코드에서 직접 읽어야 하는 경우에는 `agent.i18n`을 읽는 대신 동일한 프롬프트 파일로 `crewai.utilities.i18n.get_i18n()`을 사용하세요.
-
 ### 예시: 기본 프롬프트 커스터마이징
 
 수정하고 싶은 프롬프트를 포함하는 `custom_prompts.json` 파일을 생성하세요. 변경 사항만이 아니라 포함해야 하는 모든 최상위 프롬프트를 반드시 나열해야 합니다:
@@ -328,4 +314,4 @@ CrewAI에서의 저수준 prompt 커스터마이제이션은 매우 맞춤화되
 
 <Check>
 이제 CrewAI에서 고급 prompt 커스터마이징을 위한 기초를 갖추었습니다. 모델별 구조나 도메인별 제약에 맞춰 적용하든, 이러한 저수준 접근 방식은 agent 상호작용을 매우 전문적으로 조정할 수 있게 해줍니다.
-</Check>
+</Check>

docs/pt-BR/changelog.mdx

@@ -4,38 +4,6 @@ description: "Atualizações de produto, melhorias e correções do CrewAI"
 icon: "clock"
 mode: "wide"
 ---
-<Update label="10 jun 2026">
-  ## v1.14.7rc1
-
-  [Ver release no GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.14.7rc1)
-
-  ## O que Mudou
-
-  ### Recursos
-  - Adicionar `reset_runtime_state` para liberar o estado acumulado do barramento
-  - Lidar com suporte a ambos os prompts personalizados
-  - Desacoplar a lógica de conversa do tempo de execução e adicionar uma `conversational_definition`
-
-  ### Correções de Bugs
-  - Corrigir o escopo do estado de tempo de execução por execução para limitar o crescimento e isolar execuções concorrentes
-  - Corrigir a configuração de telemetria em `crewai-login`
-  - Corrigir o respeito a `suppress_flow_events` para eventos de execução de método
-
-  ### Documentação
-  - Atualizar imagens do OpenTelemetry
-  - Atualizar a documentação para refletir o novo estado do coletor OpenTelemetry
-  - Atualizar o changelog e a versão para v1.14.7a4
-
-  ### Refatoração
-  - Simplificar a avaliação da condição de fluxo para ser sem estado por evento
-  - Melhorar o ciclo de roteamento de conversas com uma rota a menos
-
-  ## Contribuidores
-
-  @greysonlalonde, @lorenzejay, @lucasgomide, @vinibrsl
-
-</Update>
-
 <Update label="09 jun 2026">
   ## v1.14.7a4
 

docs/pt-BR/guides/advanced/customizing-prompts.mdx

@@ -161,18 +161,6 @@ crew = Crew(
 )
 ```
 
-<Note>
-`agent.i18n` é mantido apenas para compatibilidade retroativa e está obsoleto. Para customização de prompts em tempo de execução, passe `prompt_file` para `Crew`. Para acesso programático aos slices de prompt, use diretamente o utilitário de i18n:
-</Note>
-
-```python
-from crewai.utilities.i18n import get_i18n
-
-i18n = get_i18n("custom_prompts.json")
-format_slice = i18n.slice("format")
-tool_prompt = i18n.tools("ask_question")
-```
-
 #### Opção 3: Desativar Prompts de Sistema para Modelos o1
 ```python
 agent = Agent(
@@ -220,8 +208,6 @@ Uma abordagem direta é criar um arquivo JSON para os prompts que deseja sobresc
 
 O CrewAI então mescla suas customizações com os padrões, assim você não precisa redefinir todos os prompts. Veja como:
 
-Para código que precisa ler slices de prompt diretamente, use `crewai.utilities.i18n.get_i18n()` com o mesmo arquivo de prompts em vez de ler `agent.i18n`.
-
 ### Exemplo: Customização Básica de Prompt
 
 Crie um arquivo `custom_prompts.json` com os prompts que deseja modificar. Certifique-se de listar todos os prompts de nível superior que ele deve conter, não apenas suas alterações:

lib/cli/pyproject.toml

@@ -8,7 +8,7 @@ authors = [
 ]
 requires-python = ">=3.10, <3.14"
 dependencies = [
-    "crewai-core==1.14.7rc1",
+    "crewai-core==1.14.7a4",
     "click>=8.1.7,<9",
     "pydantic>=2.11.9,<2.13",
     "pydantic-settings~=2.10.1",

lib/cli/src/crewai_cli/__init__.py

@@ -1 +1 @@
-__version__ = "1.14.7rc1"
+__version__ = "1.14.7a4"

lib/cli/src/crewai_cli/templates/crew/pyproject.toml

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
 authors = [{ name = "Your Name", email = "you@example.com" }]
 requires-python = ">=3.10,<3.14"
 dependencies = [
-    "crewai[tools]==1.14.7rc1"
+    "crewai[tools]==1.14.7a4"
 ]
 
 [project.scripts]

lib/cli/src/crewai_cli/templates/flow/pyproject.toml

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
 authors = [{ name = "Your Name", email = "you@example.com" }]
 requires-python = ">=3.10,<3.14"
 dependencies = [
-    "crewai[tools]==1.14.7rc1"
+    "crewai[tools]==1.14.7a4"
 ]
 
 [project.scripts]

lib/cli/src/crewai_cli/templates/tool/pyproject.toml

@@ -5,7 +5,7 @@ description = "Power up your crews with {{folder_name}}"
 readme = "README.md"
 requires-python = ">=3.10,<3.14"
 dependencies = [
-    "crewai[tools]==1.14.7rc1"
+    "crewai[tools]==1.14.7a4"
 ]
 
 [tool.crewai]

lib/crewai-core/src/crewai_core/__init__.py

@@ -1 +1 @@
-__version__ = "1.14.7rc1"
+__version__ = "1.14.7a4"

lib/crewai-core/src/crewai_core/telemetry.py

@@ -17,7 +17,7 @@ import contextlib
 import logging
 import os
 import threading
-from typing import Any, ClassVar, Final
+from typing import Any, Final
 
 from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
@@ -27,7 +27,7 @@ from opentelemetry.sdk.trace.export import (
     BatchSpanProcessor,
     SpanExportResult,
 )
-from opentelemetry.trace import ProxyTracerProvider, Span, Status, StatusCode
+from opentelemetry.trace import Span, Status, StatusCode
 from typing_extensions import Self
 
 
@@ -72,8 +72,8 @@ class Telemetry:
     and event-bus signal handlers (see ``crewai.telemetry.telemetry``).
     """
 
-    _instance: ClassVar[Self | None] = None
-    _lock: ClassVar[threading.Lock] = threading.Lock()
+    _instance = None
+    _lock = threading.Lock()
 
     def __new__(cls) -> Self:
         if cls._instance is None:
@@ -149,10 +149,6 @@ class Telemetry:
         if self.ready and not self.trace_set:
             try:
                 with suppress_warnings():
-                    existing_provider = trace.get_tracer_provider()
-                    if not isinstance(existing_provider, ProxyTracerProvider):
-                        self.trace_set = True
-                        return
                     trace.set_tracer_provider(self.provider)
                     self.trace_set = True
             except Exception as e:

lib/crewai-core/tests/test_smoke.py

@@ -14,7 +14,6 @@ from crewai_core import (
     version,
 )
 import pytest
-from opentelemetry.sdk.trace import TracerProvider
 
 
 def test_version_returns_string() -> None:
@@ -95,36 +94,3 @@ def test_user_data_decline_blocks(
 def test_unused_var_warning_silenced() -> None:
     # Touch os to keep the import (used by env-var fixtures above)
     assert os.environ is not None
-
-
-def test_core_telemetry_skips_duplicate_tracer_provider(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    from crewai_core.telemetry import Telemetry
-
-    Telemetry._instance = None
-    monkeypatch.delenv("OTEL_SDK_DISABLED", raising=False)
-    monkeypatch.delenv("CREWAI_DISABLE_TELEMETRY", raising=False)
-    monkeypatch.delenv("CREWAI_DISABLE_TRACKING", raising=False)
-
-    monkeypatch.setattr(
-        "crewai_core.telemetry.trace.get_tracer_provider",
-        lambda: TracerProvider(),
-    )
-
-    called = False
-
-    def fail_if_called(provider: object) -> None:
-        nonlocal called
-        called = True
-
-    monkeypatch.setattr(
-        "crewai_core.telemetry.trace.set_tracer_provider",
-        fail_if_called,
-    )
-
-    telemetry = Telemetry()
-    telemetry.set_tracer()
-
-    assert called is False
-    assert telemetry.trace_set is True

lib/crewai-files/src/crewai_files/__init__.py

@@ -152,4 +152,4 @@ __all__ = [
     "wrap_file_source",
 ]
 
-__version__ = "1.14.7rc1"
+__version__ = "1.14.7a4"

lib/crewai-tools/pyproject.toml

@@ -10,7 +10,7 @@ requires-python = ">=3.10, <3.14"
 dependencies = [
     "pytube~=15.0.0",
     "requests>=2.33.0,<3",
-    "crewai==1.14.7rc1",
+    "crewai==1.14.7a4",
     "tiktoken>=0.8.0,<0.13",
     "beautifulsoup4~=4.13.4",
     "python-docx~=1.2.0",

lib/crewai-tools/src/crewai_tools/__init__.py

@@ -330,4 +330,4 @@ __all__ = [
     "ZapierActionTools",
 ]
 
-__version__ = "1.14.7rc1"
+__version__ = "1.14.7a4"

lib/crewai/pyproject.toml

@@ -8,8 +8,8 @@ authors = [
 ]
 requires-python = ">=3.10, <3.14"
 dependencies = [
-    "crewai-core==1.14.7rc1",
-    "crewai-cli==1.14.7rc1",
+    "crewai-core==1.14.7a4",
+    "crewai-cli==1.14.7a4",
     # Core Dependencies
     "pydantic>=2.11.9,<2.13",
     "openai>=2.30.0,<3",
@@ -54,7 +54,7 @@ Repository = "https://github.com/crewAIInc/crewAI"
 
 [project.optional-dependencies]
 tools = [
-    "crewai-tools==1.14.7rc1",
+    "crewai-tools==1.14.7a4",
 ]
 embeddings = [
     "tiktoken>=0.8.0,<0.13"

lib/crewai/src/crewai/__init__.py

@@ -48,7 +48,7 @@ def _suppress_pydantic_deprecation_warnings() -> None:
 
 _suppress_pydantic_deprecation_warnings()
 
-__version__ = "1.14.7rc1"
+__version__ = "1.14.7a4"
 
 _LAZY_IMPORTS: dict[str, tuple[str, str]] = {
     "Memory": ("crewai.memory.unified_memory", "Memory"),

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -46,7 +46,6 @@ from crewai.state.checkpoint_config import CheckpointConfig, _coerce_checkpoint
 from crewai.tools.base_tool import BaseTool, Tool
 from crewai.types.callback import SerializableCallable
 from crewai.utilities.config import process_config
-from crewai.utilities.i18n import I18N, get_i18n
 from crewai.utilities.logger import Logger
 from crewai.utilities.rpm_controller import RPMController
 from crewai.utilities.string_utils import interpolate_only
@@ -187,7 +186,6 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
         tools (list[Any] | None): Tools at the agent's disposal.
         max_iter (int): Maximum iterations for an agent to execute a task.
         agent_executor: An instance of the CrewAgentExecutor class.
-        i18n (I18N): Internationalization settings.
         llm (Any): Language model that will run the agent.
         crew (Any): Crew to which the agent belongs.
 
@@ -267,14 +265,6 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
             _serialize_executor_ref, return_type=dict | None, when_used="json"
         ),
     ] = Field(default=None, description="An instance of the CrewAgentExecutor class.")
-    i18n: I18N = Field(
-        default_factory=get_i18n,
-        description="Internationalization settings.",
-        deprecated=(
-            "Agent.i18n is deprecated and will be removed in a future release. "
-            "Use crewai.utilities.i18n.get_i18n() or Crew(prompt_file=...) instead."
-        ),
-    )
 
     llm: Annotated[
         str | BaseLLM | None,

lib/crewai/src/crewai/context.py

@@ -117,10 +117,8 @@ def capture_execution_context(
     )
 
 
-def apply_execution_context(ctx: ExecutionContext | dict[str, Any]) -> None:
+def apply_execution_context(ctx: ExecutionContext) -> None:
     """Write an ExecutionContext back into the ContextVars."""
-    if isinstance(ctx, dict):
-        ctx = ExecutionContext.model_validate(ctx)
     _current_task_id.set(ctx.current_task_id)
     current_flow_request_id.set(ctx.flow_request_id)
     current_flow_id.set(ctx.flow_id)

lib/crewai/src/crewai/crew.py

@@ -1013,7 +1013,6 @@ class Crew(FlowTrackable, BaseModel):
         )
         token = attach(baggage_ctx)
 
-        runtime_scope = crewai_event_bus._enter_runtime_scope()
         try:
             inputs = prepare_kickoff(self, inputs, input_files)
 
@@ -1049,7 +1048,6 @@ class Crew(FlowTrackable, BaseModel):
                 self._memory.drain_writes()
             clear_files(self.id)
             detach(token)
-            crewai_event_bus._exit_runtime_scope(runtime_scope)
 
     def _post_kickoff(self, result: CrewOutput) -> CrewOutput:
         return result
@@ -1225,7 +1223,6 @@ class Crew(FlowTrackable, BaseModel):
         )
         token = attach(baggage_ctx)
 
-        runtime_scope = crewai_event_bus._enter_runtime_scope()
         try:
             inputs = prepare_kickoff(self, inputs, input_files)
 
@@ -1259,7 +1256,6 @@ class Crew(FlowTrackable, BaseModel):
         finally:
             clear_files(self.id)
             detach(token)
-            crewai_event_bus._exit_runtime_scope(runtime_scope)
 
     async def akickoff_for_each(
         self,

lib/crewai/src/crewai/events/event_bus.py

@@ -80,17 +80,6 @@ def is_replaying() -> bool:
     return _replaying.get()
 
 
-_runtime_state_var: contextvars.ContextVar[RuntimeState | None] = (
-    contextvars.ContextVar("crewai_runtime_state", default=None)
-)
-_registered_entity_ids_var: contextvars.ContextVar[set[int] | None] = (
-    contextvars.ContextVar("crewai_registered_entity_ids", default=None)
-)
-_runtime_scope_depth: contextvars.ContextVar[int] = contextvars.ContextVar(
-    "crewai_runtime_scope_depth", default=0
-)
-
-
 class CrewAIEventsBus:
     """Singleton event bus for handling events in CrewAI.
 
@@ -127,6 +116,7 @@ class CrewAIEventsBus:
     _futures_lock: threading.Lock
     _executor_initialized: bool
     _has_pending_events: bool
+    _runtime_state: RuntimeState | None
 
     def __new__(cls) -> Self:
         """Create or return the singleton instance.
@@ -161,6 +151,8 @@ class CrewAIEventsBus:
         self._console = ConsoleFormatter()
         self._executor_initialized = False
         self._has_pending_events = False
+        self._runtime_state: RuntimeState | None = None
+        self._registered_entity_ids: set[int] = set()
 
     def _ensure_executor_initialized(self) -> None:
         """Lazily initialize the thread pool executor and event loop.
@@ -289,51 +281,6 @@ class CrewAIEventsBus:
         """The RuntimeState currently attached to the bus, if any."""
         return self._runtime_state
 
-    @property
-    def _runtime_state(self) -> RuntimeState | None:
-        return _runtime_state_var.get()
-
-    @_runtime_state.setter
-    def _runtime_state(self, value: RuntimeState | None) -> None:
-        _runtime_state_var.set(value)
-
-    @property
-    def _registered_entity_ids(self) -> set[int]:
-        ids = _registered_entity_ids_var.get()
-        if ids is None:
-            ids = set()
-            _registered_entity_ids_var.set(ids)
-        return ids
-
-    @_registered_entity_ids.setter
-    def _registered_entity_ids(self, value: set[int]) -> None:
-        _registered_entity_ids_var.set(value)
-
-    def reset_runtime_state(self) -> None:
-        """Detach the RuntimeState and clear the entity registry."""
-        self._runtime_state = None
-        self._registered_entity_ids = set()
-
-    def _enter_runtime_scope(self) -> bool:
-        depth = _runtime_scope_depth.get()
-        _runtime_scope_depth.set(depth + 1)
-        if depth != 0:
-            return False
-        if _runtime_state_var.get() is None:
-            from crewai import RuntimeState
-
-            if RuntimeState is not None:
-                _runtime_state_var.set(RuntimeState(root=[]))
-            _registered_entity_ids_var.set(set())
-        return True
-
-    def _exit_runtime_scope(self, outermost: bool) -> None:
-        depth = _runtime_scope_depth.get()
-        _runtime_scope_depth.set(depth - 1 if depth > 0 else 0)
-        if outermost:
-            _runtime_state_var.set(None)
-            _registered_entity_ids_var.set(None)
-
     def register_entity(self, entity: Any) -> None:
         """Add an entity to the RuntimeState, creating it if needed.
 
@@ -402,7 +349,6 @@ class CrewAIEventsBus:
         source: Any,
         event: BaseEvent,
         handlers: SyncHandlerSet,
-        state: RuntimeState | None,
     ) -> None:
         """Call provided synchronous handlers.
 
@@ -410,8 +356,8 @@ class CrewAIEventsBus:
             source: The emitting object
             event: The event instance
             handlers: Frozenset of sync handlers to call
-            state: The RuntimeState captured on the emitting context
         """
+        state = self._runtime_state
         errors: list[tuple[SyncHandler, Exception]] = [
             (handler, error)
             for handler in handlers
@@ -430,7 +376,6 @@ class CrewAIEventsBus:
         source: Any,
         event: BaseEvent,
         handlers: AsyncHandlerSet,
-        state: RuntimeState | None,
     ) -> None:
         """Asynchronously call provided async handlers.
 
@@ -438,8 +383,8 @@ class CrewAIEventsBus:
             source: The object that emitted the event
             event: The event instance
             handlers: Frozenset of async handlers to call
-            state: The RuntimeState captured on the emitting context
         """
+        state = self._runtime_state
 
         async def _call(handler: AsyncHandler) -> Any:
             if _get_param_count(handler) >= 3:
@@ -454,9 +399,7 @@ class CrewAIEventsBus:
                     f"[CrewAIEventsBus] Async handler error in {getattr(handler, '__name__', handler)}: {result}"
                 )
 
-    async def _emit_with_dependencies(
-        self, source: Any, event: BaseEvent, state: RuntimeState | None
-    ) -> None:
+    async def _emit_with_dependencies(self, source: Any, event: BaseEvent) -> None:
         """Emit an event with dependency-aware handler execution.
 
         Handlers are grouped into execution levels based on their dependencies.
@@ -507,18 +450,18 @@ class CrewAIEventsBus:
 
             if level_sync:
                 if event_type is LLMStreamChunkEvent:
-                    self._call_handlers(source, event, level_sync, state)
+                    self._call_handlers(source, event, level_sync)
                 else:
                     ctx = contextvars.copy_context()
                     future = self._sync_executor.submit(
-                        ctx.run, self._call_handlers, source, event, level_sync, state
+                        ctx.run, self._call_handlers, source, event, level_sync
                     )
                     await asyncio.get_running_loop().run_in_executor(
                         None, future.result
                     )
 
             if level_async:
-                await self._acall_handlers(source, event, level_async, state)
+                await self._acall_handlers(source, event, level_async)
 
     def _register_source(self, source: Any) -> None:
         """Register the source entity in RuntimeState if applicable."""
@@ -613,23 +556,21 @@ class CrewAIEventsBus:
         self._ensure_executor_initialized()
         self._has_pending_events = True
 
-        state = self._runtime_state
-
         if has_dependencies:
             return self._track_future(
                 asyncio.run_coroutine_threadsafe(
-                    self._emit_with_dependencies(source, event, state),
+                    self._emit_with_dependencies(source, event),
                     self._loop,
                 )
             )
 
         if sync_handlers:
             if event_type is LLMStreamChunkEvent:
-                self._call_handlers(source, event, sync_handlers, state)
+                self._call_handlers(source, event, sync_handlers)
             else:
                 ctx = contextvars.copy_context()
                 sync_future = self._sync_executor.submit(
-                    ctx.run, self._call_handlers, source, event, sync_handlers, state
+                    ctx.run, self._call_handlers, source, event, sync_handlers
                 )
                 if not async_handlers:
                     return self._track_future(sync_future)
@@ -637,7 +578,7 @@ class CrewAIEventsBus:
         if async_handlers:
             return self._track_future(
                 asyncio.run_coroutine_threadsafe(
-                    self._acall_handlers(source, event, async_handlers, state),
+                    self._acall_handlers(source, event, async_handlers),
                     self._loop,
                 )
             )
@@ -649,22 +590,21 @@ class CrewAIEventsBus:
         source: Any,
         event: BaseEvent,
         handlers: AsyncHandlerSet,
-        state: RuntimeState | None,
     ) -> None:
         """Call async handlers with the replaying flag set on the loop thread."""
         token = _replaying.set(True)
         try:
-            await self._acall_handlers(source, event, handlers, state)
+            await self._acall_handlers(source, event, handlers)
         finally:
             _replaying.reset(token)
 
     async def _emit_with_dependencies_replaying(
-        self, source: Any, event: BaseEvent, state: RuntimeState | None
+        self, source: Any, event: BaseEvent
     ) -> None:
         """Dependency-aware dispatch with the replaying flag set."""
         token = _replaying.set(True)
         try:
-            await self._emit_with_dependencies(source, event, state)
+            await self._emit_with_dependencies(source, event)
         finally:
             _replaying.reset(token)
 
@@ -698,13 +638,12 @@ class CrewAIEventsBus:
         self._ensure_executor_initialized()
         self._has_pending_events = True
 
-        state = self._runtime_state
         token = _replaying.set(True)
         try:
             if has_dependencies:
                 return self._track_future(
                     asyncio.run_coroutine_threadsafe(
-                        self._emit_with_dependencies_replaying(source, event, state),
+                        self._emit_with_dependencies_replaying(source, event),
                         self._loop,
                     )
                 )
@@ -712,7 +651,7 @@ class CrewAIEventsBus:
             if sync_handlers:
                 ctx = contextvars.copy_context()
                 sync_future = self._sync_executor.submit(
-                    ctx.run, self._call_handlers, source, event, sync_handlers, state
+                    ctx.run, self._call_handlers, source, event, sync_handlers
                 )
                 self._track_future(sync_future)
                 if not async_handlers:
@@ -720,9 +659,7 @@ class CrewAIEventsBus:
 
             return self._track_future(
                 asyncio.run_coroutine_threadsafe(
-                    self._acall_handlers_replaying(
-                        source, event, async_handlers, state
-                    ),
+                    self._acall_handlers_replaying(source, event, async_handlers),
                     self._loop,
                 )
             )
@@ -790,9 +727,7 @@ class CrewAIEventsBus:
             async_handlers = self._async_handlers.get(event_type, frozenset())
 
         if async_handlers:
-            await self._acall_handlers(
-                source, event, async_handlers, self._runtime_state
-            )
+            await self._acall_handlers(source, event, async_handlers)
 
     def register_handler(
         self,

lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py

@@ -292,7 +292,7 @@ class TraceCollectionListener(BaseEventListener):
         @event_bus.on(CrewKickoffCompletedEvent)
         def on_crew_completed(source: Any, event: CrewKickoffCompletedEvent) -> None:
             self._handle_trace_event("crew_kickoff_completed", source, event)
-            if self._should_defer_session_finalization():
+            if self.batch_manager.defer_session_finalization:
                 return
             if self._nested_in_flow_execution():
                 return
@@ -306,7 +306,7 @@ class TraceCollectionListener(BaseEventListener):
         @event_bus.on(CrewKickoffFailedEvent)
         def on_crew_failed(source: Any, event: CrewKickoffFailedEvent) -> None:
             self._handle_trace_event("crew_kickoff_failed", source, event)
-            if self._should_defer_session_finalization():
+            if self.batch_manager.defer_session_finalization:
                 return
             if self._nested_in_flow_execution():
                 return
@@ -734,7 +734,7 @@ class TraceCollectionListener(BaseEventListener):
             if not self.batch_manager.is_batch_initialized():
                 return
             # Multi-turn flows defer batch finalization to finalize_session_traces().
-            if self._should_defer_session_finalization():
+            if self.batch_manager.defer_session_finalization:
                 return
             self.batch_manager.finalize_batch()
 
@@ -745,15 +745,6 @@ class TraceCollectionListener(BaseEventListener):
 
         return current_flow_id.get() is not None
 
-    def _should_defer_session_finalization(self) -> bool:
-        """True when the active trace belongs to a deferred flow session."""
-        from crewai.flow.flow_context import current_flow_defer_trace_finalization
-
-        return (
-            self.batch_manager.defer_session_finalization
-            or current_flow_defer_trace_finalization.get()
-        )
-
     def _flow_owns_trace_batch(self) -> bool:
         """True when an in-flight conversational flow already owns the trace batch."""
         if self.batch_manager.batch_owner_type == "flow":
@@ -795,11 +786,7 @@ class TraceCollectionListener(BaseEventListener):
         (``current_flow_id``) to keep LLM/tool events from falling back to an
         implicit crew batch.
         """
-        from crewai.flow.flow_context import (
-            current_flow_defer_trace_finalization,
-            current_flow_id,
-            current_flow_name,
-        )
+        from crewai.flow.flow_context import current_flow_id, current_flow_name
 
         flow_id = current_flow_id.get()
         if flow_id is None:
@@ -815,8 +802,6 @@ class TraceCollectionListener(BaseEventListener):
         }
         self.batch_manager.batch_owner_type = "flow"
         self.batch_manager.batch_owner_id = flow_id
-        if current_flow_defer_trace_finalization.get():
-            self.batch_manager.defer_session_finalization = True
         self._initialize_batch(user_context, execution_metadata)
         return True
 

lib/crewai/src/crewai/events/types/flow_events.py

@@ -1,6 +1,6 @@
 from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, field_serializer
+from pydantic import BaseModel, ConfigDict
 
 from crewai.events.base_events import BaseEvent
 
@@ -57,10 +57,6 @@ class MethodExecutionFailedEvent(FlowEvent):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    @field_serializer("error")
-    def _serialize_error(self, error: Exception) -> str:
-        return str(error)
-
 
 class MethodExecutionPausedEvent(FlowEvent):
     """Event emitted when a flow method is paused waiting for human feedback.

lib/crewai/src/crewai/experimental/conversational_mixin.py

@@ -46,9 +46,7 @@ from crewai.flow.conversation import (
     get_conversation_messages,
     receive_user_message as _receive_user_message,
 )
-from crewai.flow.dsl import listen, start
-from crewai.flow.dsl._utils import _set_flow_method_definition
-from crewai.flow.flow_definition import FlowMethodDefinition
+from crewai.flow.dsl import listen, router, start
 from crewai.utilities.types import LLMMessage
 
 
@@ -74,15 +72,6 @@ def _iter_condition_labels(condition: Any) -> set[str]:
     return set()
 
 
-def _conversation_start_router(func: Callable[..., Any]) -> Any:
-    wrapper = start()(func)
-    _set_flow_method_definition(
-        cast(Any, wrapper),
-        FlowMethodDefinition(start=True, router=True),
-    )
-    return wrapper
-
-
 class _ConversationalMixin:
     """Experimental conversational graph for ``Flow``.
 
@@ -96,7 +85,10 @@ class _ConversationalMixin:
     conversational: ClassVar[bool] = False
     conversational_config: ClassVar[ConversationConfig | None] = None
     builtin_routes: ClassVar[tuple[str, ...]] = ("converse", "end")
-    internal_routes: ClassVar[tuple[str, ...]] = ("answer_from_history",)
+    internal_routes: ClassVar[tuple[str, ...]] = (
+        "answer_from_history",
+        "conversation_start",
+    )
     builtin_route_descriptions: ClassVar[dict[str, str]] = {
         "converse": (
             "Ordinary chat, follow-ups, summaries, clarifications, and "
@@ -146,24 +138,23 @@ class _ConversationalMixin:
         def kickoff(self, *args: Any, **kwargs: Any) -> Any:
             pass
 
+    @start()
+    @_conversational_only
     def conversation_start(self) -> str | None:
-        """Return the current user message for conversational route selection.
+        """Internal Flow entrypoint that hands the user message to the router.
 
-        This remains as a plain overridable helper for compatibility. It is not
-        registered as a Flow method; ``route_conversation`` is the synthetic
-        built-in start/router that begins a conversational turn.
+        In conversational mode, ``Flow.kickoff_async`` runs all ``@start``
+        methods sequentially and this one is registered last, so any user
+        ``@start`` methods (e.g. permission loading) have already finished
+        before the returned value triggers ``route_conversation``.
         """
         state = cast(ConversationState, self.state)
         return state.current_user_message
 
-    @_conversation_start_router
+    @router(conversation_start)
     @_conversational_only
     def route_conversation(self) -> str:
         """Route the current turn to a listener label."""
-        if "conversation_start" not in {
-            str(method_name) for method_name in self._completed_methods
-        }:
-            self.conversation_start()
         state = cast(ConversationState, self.state)
         context = self.build_router_context()
         previous_intent = state.last_intent
@@ -660,16 +651,16 @@ class _ConversationalMixin:
         if not type(self)._is_conversational():
             return start_methods, False
 
-        route_conversation = "route_conversation"
-        if route_conversation not in {str(method) for method in start_methods}:
+        conversation_start = "conversation_start"
+        if conversation_start not in {str(method) for method in start_methods}:
             return start_methods, False
 
         ordered_starts = [
-            method for method in start_methods if str(method) != route_conversation
+            method for method in start_methods if str(method) != conversation_start
         ]
         ordered_starts.append(
             next(
-                method for method in start_methods if str(method) == route_conversation
+                method for method in start_methods if str(method) == conversation_start
             )
         )
         return ordered_starts, True
@@ -1056,15 +1047,12 @@ class _ConversationalMixin:
 
         trace_listener = TraceCollectionListener()
         batch_manager = trace_listener.batch_manager
-        try:
-            if batch_manager.batch_owner_type == "flow":
-                if trace_listener.first_time_handler.is_first_time:
-                    trace_listener.first_time_handler.mark_events_collected()
-                    trace_listener.first_time_handler.handle_execution_completion()
-                else:
-                    batch_manager.finalize_batch()
-        finally:
-            batch_manager.defer_session_finalization = False
+        if batch_manager.batch_owner_type == "flow":
+            if trace_listener.first_time_handler.is_first_time:
+                trace_listener.first_time_handler.mark_events_collected()
+                trace_listener.first_time_handler.handle_execution_completion()
+            else:
+                batch_manager.finalize_batch()
 
 
 __all__ = ["_ConversationalMixin"]

lib/crewai/src/crewai/flow/conversational_definition.py

@@ -39,7 +39,9 @@ class FlowConversationalDefinition(BaseModel):
     visible_agent_outputs: list[str] | Literal["all"] | None = None
     defer_trace_finalization: bool = True
     builtin_routes: list[str] = Field(default_factory=lambda: ["converse", "end"])
-    internal_routes: list[str] = Field(default_factory=lambda: ["answer_from_history"])
+    internal_routes: list[str] = Field(
+        default_factory=lambda: ["answer_from_history", "conversation_start"]
+    )
 
 
 __all__ = [

lib/crewai/src/crewai/flow/dsl/_utils.py

@@ -313,7 +313,7 @@ def _build_conversational_definition(
     internal_routes = getattr(
         flow_class,
         "internal_routes",
-        ("answer_from_history",),
+        ("answer_from_history", "conversation_start"),
     )
     if config is None:
         return FlowConversationalDefinition(
@@ -377,6 +377,11 @@ def _build_method_definition(
     else:
         method_definition = fragment.model_copy(deep=True)
 
+    # Skip <locals>/<lambda> qualnames: they can never be re-imported, so a
+    # missing handler is more honest than a dead reference.
+    if "<" not in method.__qualname__:
+        method_definition.handler = f"{method.__module__}:{method.__qualname__}"
+
     human_feedback = _build_human_feedback_definition(
         method, diagnostics, f"{path}.human_feedback"
     )

lib/crewai/src/crewai/flow/flow_context.py

@@ -15,10 +15,6 @@ current_flow_id: contextvars.ContextVar[str | None] = contextvars.ContextVar(
     "flow_id", default=None
 )
 
-current_flow_defer_trace_finalization: contextvars.ContextVar[bool] = (
-    contextvars.ContextVar("flow_defer_trace_finalization", default=False)
-)
-
 current_flow_method_name: contextvars.ContextVar[str] = contextvars.ContextVar(
     "flow_method_name", default="unknown"
 )

lib/crewai/src/crewai/flow/flow_definition.py

@@ -52,8 +52,9 @@ class FlowDefinitionDiagnostic(BaseModel):
 class FlowStateDefinition(BaseModel):
     """Static description of a Flow state contract."""
 
-    type: TypingLiteral["dict", "pydantic", "unknown"] = "dict"
+    type: TypingLiteral["dict", "pydantic", "json_schema", "unknown"] = "dict"
     ref: str | None = None
+    json_schema: dict[str, Any] | None = None
     default: Any = None
 
 
@@ -93,6 +94,7 @@ class FlowHumanFeedbackDefinition(BaseModel):
 class FlowMethodDefinition(BaseModel):
     """Static definition of one Flow method and its execution roles."""
 
+    handler: str | None = None
     start: bool | FlowDefinitionCondition | None = None
     listen: FlowDefinitionCondition | None = None
     router: bool = False

lib/crewai/src/crewai/flow/runtime.py

@@ -22,6 +22,7 @@ from concurrent.futures import Future, ThreadPoolExecutor
 import contextvars
 import copy
 import enum
+import importlib
 import inspect
 import logging
 import threading
@@ -85,16 +86,12 @@ from crewai.events.types.flow_events import (
     MethodExecutionStartedEvent,
 )
 from crewai.flow.dsl._utils import build_flow_definition
-from crewai.flow.flow_context import (
-    current_flow_defer_trace_finalization,
-    current_flow_id,
-    current_flow_name,
-    current_flow_request_id,
-)
+from crewai.flow.flow_context import current_flow_id, current_flow_request_id
 from crewai.flow.flow_definition import (
     FlowDefinition,
     FlowDefinitionCondition,
     FlowMethodDefinition,
+    FlowStateDefinition,
 )
 from crewai.flow.flow_wrappers import (
     FlowMethod,
@@ -169,6 +166,72 @@ def _condition_satisfied(condition: FlowDefinitionCondition, events: set[str]) -
     return combine(_condition_satisfied(branch, events) for branch in branches)
 
 
+def _resolve_handler(ref: str) -> Callable[..., Any]:
+    module_name, separator, qualname = ref.partition(":")
+    if not separator or not module_name or not qualname:
+        raise ValueError(
+            f"invalid handler reference {ref!r}; expected 'module:qualname'"
+        )
+    module = importlib.import_module(module_name)
+    target: Any = module
+    for part in qualname.split("."):
+        target = getattr(target, part)
+    if not callable(target):
+        raise TypeError(
+            f"handler reference {ref!r} resolved to a non-callable "
+            f"{type(target).__name__}"
+        )
+    return cast(Callable[..., Any], target)
+
+
+def _build_definition_state_model(
+    state_definition: FlowStateDefinition,
+) -> BaseModel | None:
+    kwargs = (
+        dict(state_definition.default)
+        if isinstance(state_definition.default, dict)
+        else {}
+    )
+
+    model_class: type[BaseModel] | None = None
+    if state_definition.ref:
+        try:
+            resolved = _resolve_handler(state_definition.ref)
+        except Exception:
+            logger.warning(
+                "Could not import state ref %r", state_definition.ref, exc_info=True
+            )
+        else:
+            if isinstance(resolved, type) and issubclass(resolved, BaseModel):
+                model_class = resolved
+            else:
+                logger.warning(
+                    "State ref %r is not a pydantic model", state_definition.ref
+                )
+
+    if model_class is None and state_definition.json_schema:
+        from crewai.utilities.pydantic_schema_utils import create_model_from_schema
+
+        try:
+            model_class = create_model_from_schema(state_definition.json_schema)
+        except Exception:
+            logger.warning(
+                "Could not build a state model from the declared json_schema",
+                exc_info=True,
+            )
+
+    if model_class is None:
+        return None
+
+    if not issubclass(model_class, FlowState):
+
+        class StateWithId(FlowState, model_class):  # type: ignore[misc, valid-type]
+            pass
+
+        model_class = StateWithId
+    return model_class(**kwargs)
+
+
 def _iter_condition_events(condition: FlowDefinitionCondition) -> Iterator[str]:
     if isinstance(condition, str):
         yield condition
@@ -695,21 +758,24 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         return flow_definition
 
     @classmethod
-    def _start_method_names(cls) -> list[FlowMethodName]:
+    def from_definition(cls, definition: FlowDefinition) -> Flow[Any]:
+        """Build a runnable Flow directly from a definition; no subclass required."""
+        return cls.model_validate({}, context={"flow_definition": definition})
+
+    def _start_method_names(self) -> list[FlowMethodName]:
         return [
             FlowMethodName(method_name)
-            for method_name, method_definition in cls.flow_definition().methods.items()
+            for method_name, method_definition in self._definition.methods.items()
             if method_definition.is_start
         ]
 
-    @classmethod
     def _listener_methods(
-        cls,
+        self,
     ) -> Iterator[tuple[FlowMethodName, FlowMethodDefinition, FlowDefinitionCondition]]:
         # (name, definition, condition) for every non-start method that listens.
         # Routers are included (they listen too); callers wanting only plain
         # listeners filter on definition.router.
-        for method_name, method_definition in cls.flow_definition().methods.items():
+        for method_name, method_definition in self._definition.methods.items():
             if method_definition.listen is not None and not method_definition.is_start:
                 yield (
                     FlowMethodName(method_name),
@@ -717,25 +783,22 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     method_definition.listen,
                 )
 
-    @classmethod
     def _start_condition(
-        cls, method_name: FlowMethodName
+        self, method_name: FlowMethodName
     ) -> FlowDefinitionCondition | None:
-        method_definition = cls.flow_definition().methods[str(method_name)]
+        method_definition = self._definition.methods[str(method_name)]
         start = method_definition.start
         if isinstance(start, (str, dict)):
             return start
         return None
 
-    @classmethod
     def _listen_condition(
-        cls, method_name: FlowMethodName
+        self, method_name: FlowMethodName
     ) -> FlowDefinitionCondition | None:
-        return cls.flow_definition().methods[str(method_name)].listen
+        return self._definition.methods[str(method_name)].listen
 
-    @classmethod
-    def _is_router(cls, method_name: FlowMethodName) -> bool:
-        return cls.flow_definition().methods[str(method_name)].router
+    def _is_router(self, method_name: FlowMethodName) -> bool:
+        return self._definition.methods[str(method_name)].router
 
     initial_state: Annotated[  # type: ignore[type-arg]
         type[BaseModel] | type[dict] | dict[str, Any] | BaseModel | None,
@@ -878,7 +941,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         restore_event_scope(())
         reset_last_event_id()
 
-    _methods: dict[FlowMethodName, FlowMethod[Any, Any]] = PrivateAttr(
+    _methods: dict[FlowMethodName, Callable[..., Any]] = PrivateAttr(
         default_factory=dict
     )
     _method_execution_counts: dict[FlowMethodName, int] = PrivateAttr(
@@ -892,6 +955,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         PrivateAttr(default=None)
     )
     _method_outputs: list[Any] = PrivateAttr(default_factory=list)
+    _definition: FlowDefinition = PrivateAttr()
     _state_lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)
     _or_listeners_lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)
     _completed_methods: set[FlowMethodName] = PrivateAttr(default_factory=set)
@@ -920,15 +984,27 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             object.__setattr__(self, name, value)
 
     def model_post_init(self, __context: Any) -> None:
-        self._flow_post_init()
+        definition = (
+            __context.get("flow_definition") if isinstance(__context, dict) else None
+        )
+        self._flow_post_init(definition)
 
-    def _flow_post_init(self) -> None:
+    def _flow_post_init(self, definition: FlowDefinition | None = None) -> None:
         """Heavy initialization: state creation, events, memory, method registration."""
         if getattr(self, "_flow_post_init_done", False):
             return
         object.__setattr__(self, "_flow_post_init_done", True)
         self._initialize_runtime_extension_attrs()
 
+        self._definition = definition or type(self).flow_definition()
+        if self.name and self.name != self._definition.name:
+            self._definition = self._definition.model_copy(update={"name": self.name})
+        methods = (
+            self._handler_bound_methods()
+            if definition is not None
+            else self._class_bound_methods()
+        )
+
         if self._state is None:
             self._state = self._create_initial_state()
 
@@ -943,7 +1019,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 self,
                 FlowCreatedEvent(
                     type="flow_created",
-                    flow_name=self.name or self.__class__.__name__,
+                    flow_name=self._definition.name,
                 ),
             )
 
@@ -953,17 +1029,44 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         if self.memory is None and not getattr(self, "_skip_auto_memory", False):
             from crewai.memory.utils import sanitize_scope_name
 
-            flow_name = sanitize_scope_name(self.name or self.__class__.__name__)
+            flow_name = sanitize_scope_name(self._definition.name)
             self.memory = Memory(root_scope=f"/flow/{flow_name}")
 
-        # Build the runtime method lookup from the static FlowDefinition.
-        for method_name in type(self).flow_definition().methods:
+        self._methods.update(methods)
+
+    def _handler_bound_methods(self) -> dict[FlowMethodName, Callable[..., Any]]:
+        methods: dict[FlowMethodName, Callable[..., Any]] = {}
+        unresolved: list[str] = []
+        for method_name, method_definition in self._definition.methods.items():
+            if method_definition.handler is None:
+                unresolved.append(f"{method_name}: no handler")
+                continue
+            try:
+                handler = _resolve_handler(method_definition.handler)
+            except Exception as e:
+                unresolved.append(f"{method_name}: {e}")
+                continue
+            if getattr(handler, "__self__", None) is None:
+                handler = handler.__get__(self, type(self))
+            methods[FlowMethodName(method_name)] = handler
+        if unresolved:
+            raise ValueError(
+                f"Cannot build flow {self._definition.name!r} from its definition; "
+                "methods with missing or unresolvable handlers: "
+                + "; ".join(unresolved)
+            )
+        return methods
+
+    def _class_bound_methods(self) -> dict[FlowMethodName, Callable[..., Any]]:
+        methods: dict[FlowMethodName, Callable[..., Any]] = {}
+        for method_name in self._definition.methods:
             method = getattr(self, method_name, None)
             if method is None:
                 continue
             if not hasattr(method, "__self__"):
-                method = method.__get__(self, self.__class__)
-            self._methods[FlowMethodName(method_name)] = method
+                method = method.__get__(self, type(self))
+            methods[FlowMethodName(method_name)] = method
+        return methods
 
     def recall(self, query: str, **kwargs: Any) -> Any:
         """Recall relevant memories. Delegates to this flow's memory.
@@ -1041,7 +1144,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
     def _start_condition_triggered_by(
         self, method_name: FlowMethodName, trigger: FlowMethodName
     ) -> bool:
-        condition = type(self)._start_condition(method_name)
+        condition = self._start_condition(method_name)
         if condition is None:
             return False
         return self._condition_met(
@@ -1069,7 +1172,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             trigger_str = str(trigger)
             to_discard: list[FlowMethodName] = []
             for listener_name in candidates:
-                condition = type(self)._listen_condition(listener_name)
+                condition = self._listen_condition(listener_name)
                 if condition is None:
                     continue
                 if trigger_str in _iter_condition_events(condition):
@@ -1091,9 +1194,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         racing_groups: dict[frozenset[FlowMethodName], FlowMethodName] = {}
         listener_conditions: dict[FlowMethodName, FlowDefinitionCondition] = {
             listener_name: condition
-            for listener_name, method_definition, condition in type(
-                self
-            )._listener_methods()
+            for listener_name, method_definition, condition in self._listener_methods()
             if not method_definition.router
         }
 
@@ -1366,7 +1467,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 self,
                 FlowStartedEvent(
                     type="flow_started",
-                    flow_name=self.name or self.__class__.__name__,
+                    flow_name=self._definition.name,
                     inputs=None,
                 ),
             )
@@ -1442,7 +1543,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 self,
                 MethodExecutionFinishedEvent(
                     type="method_execution_finished",
-                    flow_name=self.name or self.__class__.__name__,
+                    flow_name=self._definition.name,
                     method_name=context.method_name,
                     result=collapsed_outcome if emit else result,
                     state=self._state,
@@ -1496,7 +1597,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     self,
                     FlowPausedEvent(
                         type="flow_paused",
-                        flow_name=self.name or self.__class__.__name__,
+                        flow_name=self._definition.name,
                         flow_id=e.context.flow_id,
                         method_name=e.context.method_name,
                         state=self._copy_and_serialize_state(),
@@ -1519,15 +1620,12 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             )
             self._event_futures.clear()
 
-        if (
-            not self.suppress_flow_events
-            and not self._should_defer_trace_finalization()
-        ):
+        if not self.suppress_flow_events:
             future = crewai_event_bus.emit(
                 self,
                 FlowFinishedEvent(
                     type="flow_finished",
-                    flow_name=self.name or self.__class__.__name__,
+                    flow_name=self._definition.name,
                     result=final_result,
                     state=self._copy_and_serialize_state(),
                 ),
@@ -1539,12 +1637,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     logger.warning("FlowFinishedEvent handler failed", exc_info=True)
 
             trace_listener = TraceCollectionListener()
-            if (
-                trace_listener.batch_manager.batch_owner_type == "flow"
-                and current_flow_id.get() == self.flow_id
-                and not trace_listener.batch_manager.defer_session_finalization
-                and not current_flow_defer_trace_finalization.get()
-            ):
+            if trace_listener.batch_manager.batch_owner_type == "flow":
                 if trace_listener.first_time_handler.is_first_time:
                     trace_listener.first_time_handler.mark_events_collected()
                     trace_listener.first_time_handler.handle_execution_completion()
@@ -1593,7 +1686,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     return cast(T, {"id": str(uuid4())})
 
         if init_state is None:
-            return cast(T, {"id": str(uuid4())})
+            return cast(T, self._create_definition_state())
 
         if isinstance(init_state, type):
             state_class = init_state
@@ -1635,6 +1728,34 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             f"Initial state must be dict or BaseModel, got {type(self.initial_state)}"
         )
 
+    def _create_definition_state(self) -> dict[str, Any] | BaseModel:
+        state_definition = self._definition.state
+        if state_definition is None:
+            return {"id": str(uuid4())}
+        if state_definition.type in ("pydantic", "json_schema"):
+            state = _build_definition_state_model(state_definition)
+            if state is not None:
+                return state
+            logger.error(
+                "Flow %r declares %s state but neither ref nor json_schema "
+                "produced a model; falling back to dict state",
+                self._definition.name,
+                state_definition.type,
+            )
+        elif state_definition.type == "unknown":
+            logger.warning(
+                "Flow %r declares state of unknown type; falling back to dict state",
+                self._definition.name,
+            )
+        dict_state: dict[str, Any] = (
+            dict(state_definition.default)
+            if isinstance(state_definition.default, dict)
+            else {}
+        )
+        if "id" not in dict_state:
+            dict_state["id"] = str(uuid4())
+        return dict_state
+
     def _copy_state(self) -> T:
         """Create a copy of the current state.
 
@@ -1935,17 +2056,13 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 restore_from_state_id=restore_from_state_id,
             )
 
-        runtime_scope = crewai_event_bus._enter_runtime_scope()
         try:
-            try:
-                asyncio.get_running_loop()
-                ctx = contextvars.copy_context()
-                with ThreadPoolExecutor(max_workers=1) as pool:
-                    return pool.submit(ctx.run, asyncio.run, _run_flow()).result()
-            except RuntimeError:
-                return asyncio.run(_run_flow())
-        finally:
-            crewai_event_bus._exit_runtime_scope(runtime_scope)
+            asyncio.get_running_loop()
+            ctx = contextvars.copy_context()
+            with ThreadPoolExecutor(max_workers=1) as pool:
+                return pool.submit(ctx.run, asyncio.run, _run_flow()).result()
+        except RuntimeError:
+            return asyncio.run(_run_flow())
 
     async def kickoff_async(
         self,
@@ -2037,23 +2154,12 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         flow_token = attach(ctx)
 
         flow_id_token = None
-        flow_name_token = None
-        flow_defer_trace_finalization_token = None
         request_id_token = None
         if current_flow_id.get() is None:
             flow_id_token = current_flow_id.set(self.flow_id)
-            flow_name_token = current_flow_name.set(
-                self.name or self.__class__.__name__
-            )
-            flow_defer_trace_finalization_token = (
-                current_flow_defer_trace_finalization.set(
-                    self._should_defer_trace_finalization()
-                )
-            )
         if current_flow_request_id.get() is None:
             request_id_token = current_flow_request_id.set(self.flow_id)
 
-        runtime_scope = crewai_event_bus._enter_runtime_scope()
         try:
             # Reset flow state for fresh execution unless restoring from persistence
             is_restoring = (
@@ -2145,10 +2251,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             should_emit_flow_started = not (
                 defer_trace_finalization and deferred_started_event_id
             )
-            if current_flow_id.get() == self.flow_id:
-                TraceCollectionListener().batch_manager.defer_session_finalization = (
-                    defer_trace_finalization
-                )
 
             if (
                 defer_trace_finalization
@@ -2166,7 +2268,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 # explicit finalization call closes the batch.
                 started_event = FlowStartedEvent(
                     type="flow_started",
-                    flow_name=self.name or self.__class__.__name__,
+                    flow_name=self._definition.name,
                     inputs=inputs,
                 )
                 future = crewai_event_bus.emit(self, started_event)
@@ -2206,11 +2308,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 # Determine which start methods to execute at kickoff
                 # Conditional start methods are only triggered by their conditions
                 # UNLESS there are no unconditional starts (then all starts run as entry points)
-                start_methods = type(self)._start_method_names()
+                start_methods = self._start_method_names()
                 unconditional_starts = [
                     start_method
                     for start_method in start_methods
-                    if type(self)._start_condition(start_method) is None
+                    if self._start_condition(start_method) is None
                 ]
                 # If there are unconditional starts, only run those at kickoff
                 # If there are NO unconditional starts, run all starts (including conditional ones)
@@ -2258,7 +2360,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                         self,
                         FlowPausedEvent(
                             type="flow_paused",
-                            flow_name=self.name or self.__class__.__name__,
+                            flow_name=self._definition.name,
                             flow_id=e.context.flow_id,
                             method_name=e.context.method_name,
                             state=self._copy_and_serialize_state(),
@@ -2308,7 +2410,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     self,
                     FlowFinishedEvent(
                         type="flow_finished",
-                        flow_name=self.name or self.__class__.__name__,
+                        flow_name=self._definition.name,
                         result=final_output,
                         state=self._copy_and_serialize_state(),
                     ),
@@ -2322,12 +2424,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                         )
 
                 trace_listener = TraceCollectionListener()
-                if (
-                    trace_listener.batch_manager.batch_owner_type == "flow"
-                    and current_flow_id.get() == self.flow_id
-                    and not trace_listener.batch_manager.defer_session_finalization
-                    and not current_flow_defer_trace_finalization.get()
-                ):
+                if trace_listener.batch_manager.batch_owner_type == "flow":
                     if trace_listener.first_time_handler.is_first_time:
                         trace_listener.first_time_handler.mark_events_collected()
                         trace_listener.first_time_handler.handle_execution_completion()
@@ -2341,16 +2438,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 self.memory.drain_writes()
             if request_id_token is not None:
                 current_flow_request_id.reset(request_id_token)
-            if flow_defer_trace_finalization_token is not None:
-                current_flow_defer_trace_finalization.reset(
-                    flow_defer_trace_finalization_token
-                )
-            if flow_name_token is not None:
-                current_flow_name.reset(flow_name_token)
             if flow_id_token is not None:
                 current_flow_id.reset(flow_id_token)
             detach(flow_token)
-            crewai_event_bus._exit_runtime_scope(runtime_scope)
 
     async def akickoff(
         self,
@@ -2394,7 +2484,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             MethodExecutionFinishedEvent,
             MethodExecutionFailedEvent,
         )
-        flow_name = self.name or self.__class__.__name__
+        flow_name = self._definition.name
         nodes = sorted(
             (
                 n
@@ -2453,7 +2543,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         )
 
         # If start method is a router, use its result as an additional trigger
-        if type(self)._is_router(start_method_name) and result is not None:
+        if self._is_router(start_method_name) and result is not None:
             # Execute listeners for the start method name first
             await self._execute_listeners(start_method_name, result, finished_event_id)
             # Then execute listeners for the router result (e.g., "approved")
@@ -2473,15 +2563,16 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
     def _inject_trigger_payload_for_start_method(
         self, original_method: Callable[..., Any]
     ) -> Callable[..., Any]:
+        accepts_trigger_payload = (
+            "crewai_trigger_payload" in inspect.signature(original_method).parameters
+        )
+
         def prepare_kwargs(
             *args: Any, **kwargs: Any
         ) -> tuple[tuple[Any, ...], dict[str, Any]]:
             inputs = cast(dict[str, Any], baggage.get_baggage("flow_inputs") or {})
             trigger_payload = inputs.get("crewai_trigger_payload")
 
-            sig = inspect.signature(original_method)
-            accepts_trigger_payload = "crewai_trigger_payload" in sig.parameters
-
             if trigger_payload is not None and accepts_trigger_payload:
                 kwargs["crewai_trigger_payload"] = trigger_payload
             elif trigger_payload is not None:
@@ -2531,7 +2622,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     MethodExecutionStartedEvent(
                         type="method_execution_started",
                         method_name=method_name,
-                        flow_name=self.name or self.__class__.__name__,
+                        flow_name=self._definition.name,
                         params=dumped_params,
                         state=self._copy_and_serialize_state(),
                     ),
@@ -2583,7 +2674,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 finished_event = MethodExecutionFinishedEvent(
                     type="method_execution_finished",
                     method_name=method_name,
-                    flow_name=self.name or self.__class__.__name__,
+                    flow_name=self._definition.name,
                     state=self._copy_and_serialize_state(),
                     result=result,
                 )
@@ -2612,7 +2703,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                         MethodExecutionPausedEvent(
                             type="method_execution_paused",
                             method_name=method_name,
-                            flow_name=self.name or self.__class__.__name__,
+                            flow_name=self._definition.name,
                             state=self._copy_and_serialize_state(),
                             flow_id=e.context.flow_id,
                             message=e.context.message,
@@ -2628,7 +2719,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     MethodExecutionFailedEvent(
                         type="method_execution_failed",
                         method_name=method_name,
-                        flow_name=self.name or self.__class__.__name__,
+                        flow_name=self._definition.name,
                         error=e,
                     ),
                 )
@@ -2760,7 +2851,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                         await asyncio.gather(*tasks)
 
                 if current_trigger in router_results:
-                    for method_name in type(self)._start_method_names():
+                    for method_name in self._start_method_names():
                         if self._start_condition_triggered_by(
                             method_name, current_trigger
                         ):
@@ -2791,9 +2882,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
     ) -> list[FlowMethodName]:
         triggered: list[FlowMethodName] = []
 
-        for listener_name, method_definition, condition in type(
-            self
-        )._listener_methods():
+        for listener_name, method_definition, condition in self._listener_methods():
             is_router = method_definition.router
             if router_only != is_router:
                 continue
@@ -2859,10 +2948,10 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
 
                 # For routers, also check if any conditional starts they triggered are completed
                 # If so, continue their chains
-                if type(self)._is_router(listener_name):
-                    for start_method_name in type(self)._start_method_names():
+                if self._is_router(listener_name):
+                    for start_method_name in self._start_method_names():
                         if (
-                            type(self)._start_condition(start_method_name) is not None
+                            self._start_condition(start_method_name) is not None
                             and start_method_name in self._completed_methods
                         ):
                             # This conditional start was executed, continue its chain
@@ -2881,8 +2970,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             method = self._methods[listener_name]
 
             sig = inspect.signature(method)
-            params = list(sig.parameters.values())
-            method_params = [p for p in params if p.name != "self"]
+            method_params = [p for p in sig.parameters.values() if p.name != "self"]
 
             if triggering_event_id:
                 with triggered_by_scope(triggering_event_id):
@@ -3038,7 +3126,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             self,
             FlowInputRequestedEvent(
                 type="flow_input_requested",
-                flow_name=self.name or self.__class__.__name__,
+                flow_name=self._definition.name,
                 method_name=method_name,
                 message=message,
                 metadata=metadata,
@@ -3105,7 +3193,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             self,
             FlowInputReceivedEvent(
                 type="flow_input_received",
-                flow_name=self.name or self.__class__.__name__,
+                flow_name=self._definition.name,
                 method_name=method_name,
                 message=message,
                 response=response,
@@ -3143,7 +3231,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             self,
             HumanFeedbackRequestedEvent(
                 type="human_feedback_requested",
-                flow_name=self.name or self.__class__.__name__,
+                flow_name=self._definition.name,
                 method_name="",  # Will be set by decorator if needed
                 output=output,
                 message=message,
@@ -3172,7 +3260,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 self,
                 HumanFeedbackReceivedEvent(
                     type="human_feedback_received",
-                    flow_name=self.name or self.__class__.__name__,
+                    flow_name=self._definition.name,
                     method_name="",  # Will be set by decorator if needed
                     feedback=feedback,
                     outcome=None,  # Will be determined after collapsing
@@ -3347,7 +3435,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             self,
             FlowPlotEvent(
                 type="flow_plot",
-                flow_name=self.name or self.__class__.__name__,
+                flow_name=self._definition.name,
             ),
         )
         structure = build_flow_structure(cast(Any, self))

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

@@ -30,7 +30,7 @@ from opentelemetry.sdk.trace.export import (
     BatchSpanProcessor,
     SpanExportResult,
 )
-from opentelemetry.trace import ProxyTracerProvider, Span
+from opentelemetry.trace import Span
 from typing_extensions import Self
 
 from crewai.events.event_bus import crewai_event_bus
@@ -162,10 +162,6 @@ class Telemetry:
         if self.ready and not self.trace_set:
             try:
                 with suppress_warnings():
-                    existing_provider = trace.get_tracer_provider()
-                    if not isinstance(existing_provider, ProxyTracerProvider):
-                        self.trace_set = True
-                        return
                     trace.set_tracer_provider(self.provider)
                     self.trace_set = True
             except Exception as e:

lib/crewai/src/crewai/utilities/pydantic_schema_utils.py

@@ -999,7 +999,11 @@ def _json_schema_to_pydantic_field(
     if examples:
         schema_extra["examples"] = examples
 
-    default = ... if is_required else None
+    default = (
+        json_schema["default"]
+        if "default" in json_schema
+        else (... if is_required else None)
+    )
 
     if isinstance(type_, type) and issubclass(type_, (int, float)):
         if "minimum" in json_schema:

lib/crewai/tests/agents/test_agent.py

@@ -4,7 +4,6 @@ import os
 import threading
 from unittest import mock
 from unittest.mock import MagicMock, patch
-import warnings
 
 from crewai.agents.crew_agent_executor import AgentFinish, CrewAgentExecutor
 from crewai.constants import DEFAULT_LLM_MODEL
@@ -78,51 +77,6 @@ def test_agent_creation():
     assert agent.backstory == "test backstory"
 
 
-def test_agent_exposes_i18n_for_backward_compatibility():
-    from crewai.utilities.i18n import I18N_DEFAULT
-
-    agent = Agent(role="test role", goal="test goal", backstory="test backstory")
-
-    with pytest.warns(DeprecationWarning, match="Agent.i18n is deprecated"):
-        i18n = agent.i18n
-
-    assert i18n is I18N_DEFAULT
-    assert isinstance(i18n.slice("role_playing"), str)
-
-
-def test_agent_accepts_custom_i18n():
-    from crewai.utilities.i18n import I18N
-
-    prompt_file = os.path.join(
-        os.path.dirname(__file__), "..", "utilities", "prompts.json"
-    )
-    i18n = I18N(prompt_file=prompt_file)
-    agent = Agent(
-        role="test role",
-        goal="test goal",
-        backstory="test backstory",
-        i18n=i18n,
-    )
-
-    with pytest.warns(DeprecationWarning, match="Agent.i18n is deprecated"):
-        agent_i18n = agent.i18n
-
-    assert agent_i18n is i18n
-    assert agent_i18n.slice("role_playing") == "Lorem ipsum dolor sit amet"
-
-
-def test_agent_copy_does_not_emit_i18n_deprecation_warning():
-    agent = Agent(role="test role", goal="test goal", backstory="test backstory")
-
-    with warnings.catch_warnings(record=True) as caught_warnings:
-        warnings.simplefilter("always", DeprecationWarning)
-        agent.copy()
-
-    assert not any(
-        "Agent.i18n is deprecated" in str(w.message) for w in caught_warnings
-    )
-
-
 def test_agent_with_only_system_template():
     """Test that an agent with only system_template works without errors."""
     agent = Agent(

lib/crewai/tests/events/test_event_replay.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import threading
 from typing import Any
 from unittest.mock import patch
 
@@ -110,79 +109,10 @@ class TestCheckpointListenerOptsOut:
             assert do_cp.call_count == 0
 
 
-class TestCheckpointResumeReplaysEvents:
-    """A flow resumed from a checkpoint replays MethodExecution* events for
-    completed methods and executes the pending ones. The checkpoint persists
-    the event record, which is reloaded into the per-run runtime state.
+class TestFlowResumeReplaysEvents:
+    """End-to-end: a resumed flow emits MethodExecution* events for completed methods."""
 
-    ``step_c`` is gated on a threading.Event so the flow is frozen with exactly
-    ``step_a`` and ``step_b`` completed when the checkpoint is written — the
-    mid-run snapshot is deterministic rather than dependent on write timing.
-    """
-
-    def test_resume_replays_completed_and_executes_pending(self, tmp_path) -> None:
-        from crewai.flow.flow import Flow, listen, start
-        from crewai.state.checkpoint_config import CheckpointConfig
-
-        at_step_c = threading.Event()
-        release = threading.Event()
-        captured: list[Any] = []
-
-        class ThreeStepFlow(Flow[dict]):
-            @start()
-            def step_a(self) -> str:
-                return "a"
-
-            @listen(step_a)
-            def step_b(self) -> str:
-                return "b"
-
-            @listen(step_b)
-            def step_c(self) -> str:
-                captured.append(crewai_event_bus.runtime_state)
-                at_step_c.set()
-                release.wait(timeout=10)
-                return "c"
-
-        runner = threading.Thread(target=ThreeStepFlow().kickoff)
-        runner.start()
-        try:
-            assert at_step_c.wait(timeout=10)
-            location = captured[0].checkpoint(str(tmp_path / "cp"))
-        finally:
-            release.set()
-            runner.join(timeout=10)
-
-        captured_started: list[str] = []
-        captured_finished: list[str] = []
-
-        with crewai_event_bus.scoped_handlers():
-
-            @crewai_event_bus.on(MethodExecutionStartedEvent)
-            def _cs(_: Any, event: MethodExecutionStartedEvent) -> None:
-                captured_started.append(event.method_name)
-
-            @crewai_event_bus.on(MethodExecutionFinishedEvent)
-            def _cf(_: Any, event: MethodExecutionFinishedEvent) -> None:
-                captured_finished.append(event.method_name)
-
-            ThreeStepFlow().kickoff(
-                from_checkpoint=CheckpointConfig(restore_from=location)
-            )
-
-        assert captured_started == ["step_a", "step_b", "step_c"]
-        assert captured_finished == ["step_a", "step_b", "step_c"]
-
-
-class TestPersistResumeDoesNotReplayCompletedEvents:
-    """A @persist resume continues from pending methods only.
-
-    @persist stores flow state, not the event record, so completed-method
-    events have no persisted source to replay from. Runtime state is scoped
-    per run, so flow1's events are not visible to flow2.
-    """
-
-    def test_persist_resume_executes_only_pending_methods(self, tmp_path) -> None:
+    def test_resume_dispatches_completed_method_events(self, tmp_path) -> None:
         from crewai.flow.flow import Flow, listen, start
         from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
 
@@ -202,6 +132,9 @@ class TestPersistResumeDoesNotReplayCompletedEvents:
             def step_c(self) -> str:
                 return "c"
 
+        if crewai_event_bus.runtime_state is not None:
+            crewai_event_bus.runtime_state.event_record.clear()
+
         flow1 = ThreeStepFlow(persistence=persistence)
         flow1.kickoff()
         flow_id = flow1.state["id"]
@@ -224,5 +157,9 @@ class TestPersistResumeDoesNotReplayCompletedEvents:
 
             flow2.kickoff(inputs={"id": flow_id})
 
-        assert captured_started == ["step_c"]
-        assert captured_finished == ["step_c"]
+        assert captured_started.count("step_a") == 1
+        assert captured_started.count("step_b") == 1
+        assert captured_started.count("step_c") == 1
+        assert captured_finished.count("step_a") == 1
+        assert captured_finished.count("step_b") == 1
+        assert captured_finished.count("step_c") == 1

lib/crewai/tests/telemetry/test_telemetry.py

@@ -6,7 +6,6 @@ import pytest
 from crewai import Agent, Crew, Task
 from crewai.telemetry import Telemetry
 from opentelemetry import trace
-from opentelemetry.sdk.trace import TracerProvider
 
 
 @pytest.fixture(autouse=True)
@@ -54,23 +53,6 @@ def test_telemetry_enabled_by_default():
             assert telemetry.ready is True
 
 
-def test_set_tracer_skips_when_provider_already_configured():
-    """A second telemetry instance must not re-install the global provider."""
-    with (
-        patch.dict(os.environ, {}, clear=True),
-        patch(
-            "crewai.telemetry.telemetry.trace.get_tracer_provider",
-            return_value=TracerProvider(),
-        ),
-        patch("crewai.telemetry.telemetry.trace.set_tracer_provider") as mock_set,
-    ):
-        telemetry = Telemetry()
-        telemetry.set_tracer()
-
-    mock_set.assert_not_called()
-    assert telemetry.trace_set is True
-
-
 @patch("crewai.telemetry.telemetry.logger.error")
 @patch(
     "opentelemetry.exporter.otlp.proto.http.trace_exporter.OTLPSpanExporter.export",

lib/crewai/tests/test_event_record.py

@@ -409,31 +409,4 @@ class TestRuntimeStateIntegration:
             old_json, context={"from_checkpoint": True}
         )
         assert len(restored.root) == 1
-        assert len(restored.event_record) == 0
-
-    def test_reset_runtime_state_clears_state_and_registry(self):
-        from crewai import Agent, Crew, RuntimeState
-        from crewai.events.event_bus import crewai_event_bus
-
-        if RuntimeState is None:
-            pytest.skip("RuntimeState unavailable (model_rebuild failed)")
-
-        agent = Agent(role="test", goal="test", backstory="test", llm="gpt-4o-mini")
-        crew = Crew(agents=[agent], tasks=[], verbose=False)
-
-        previous_state = crewai_event_bus._runtime_state
-        previous_ids = crewai_event_bus._registered_entity_ids
-        crewai_event_bus._runtime_state = None
-        crewai_event_bus._registered_entity_ids = set()
-        try:
-            crewai_event_bus.register_entity(crew)
-            assert crewai_event_bus.runtime_state is not None
-            assert crewai_event_bus._registered_entity_ids
-
-            crewai_event_bus.reset_runtime_state()
-
-            assert crewai_event_bus.runtime_state is None
-            assert crewai_event_bus._registered_entity_ids == set()
-        finally:
-            crewai_event_bus._runtime_state = previous_state
-            crewai_event_bus._registered_entity_ids = previous_ids
+        assert len(restored.event_record) == 0

lib/crewai/tests/test_flow.py

@@ -1157,6 +1157,25 @@ def test_flow_name():
     assert flow.name == "MyFlow"
 
 
+def test_flow_custom_name_overrides_class_name_in_events():
+    class InternalFlowClass(Flow):
+        name = "PublicName"
+
+        @start()
+        def begin(self):
+            return "done"
+
+    received = []
+
+    @crewai_event_bus.on(FlowStartedEvent)
+    def handle(source, event):
+        received.append(event)
+
+    InternalFlowClass().kickoff()
+
+    assert received[0].flow_name == "PublicName"
+
+
 def test_nested_and_or_conditions():
     """Test nested conditions like or_(and_(A, B), and_(C, D)).
 

lib/crewai/tests/test_flow_conversation.py

@@ -26,11 +26,7 @@ from crewai.experimental import (
     RouterConfig,
 )
 from crewai.flow import Flow, ChatState, listen, start
-from crewai.flow.flow_context import (
-    current_flow_defer_trace_finalization,
-    current_flow_id,
-    current_flow_name,
-)
+from crewai.flow.flow_context import current_flow_id, current_flow_name
 from crewai.flow.conversation import (
     append_message,
     get_conversation_messages,
@@ -602,9 +598,9 @@ class TestConversationalFlow:
         """Conversational flows: user ``@start`` methods finish before router fires.
 
         Non-chat flows run ``@start`` methods in parallel via ``asyncio.gather``,
-        which would race with ``route_conversation`` and let the router fire
+        which would race with ``conversation_start`` and let the router fire
         before user setup finished. In conversational mode the framework runs
-        them sequentially, with ``route_conversation`` last.
+        them sequentially, with ``conversation_start`` last.
         """
         order: list[str] = []
 
@@ -647,10 +643,15 @@ class TestConversationalFlow:
         assert "attach_bus" in order  # still fires every turn
         assert "route_turn" in order
 
-    def test_subclass_can_override_conversation_start_helper(
+    def test_subclass_can_override_conversation_start_without_redecorating(
         self,
     ) -> None:
-        """The compatibility helper remains overridable without adding a Flow node."""
+        """Overriding an inherited ``@start`` method must not unregister it.
+
+        Before the metaclass fix, subclasses had to re-apply ``@start()`` on
+        every override or the parent's ``conversation_start`` would silently
+        drop out of the start registry — leaving the flow with nothing to fire.
+        """
 
         bootstrap_calls: list[str] = []
 
@@ -671,38 +672,6 @@ class TestConversationalFlow:
         flow = BootstrapFlow()
         flow.handle_turn("hi")
 
-        assert bootstrap_calls == ["ran"]
-        assert "conversation_start" not in BootstrapFlow.flow_definition().methods
-        route_definition = BootstrapFlow.flow_definition().methods["route_conversation"]
-        assert route_definition.start is True
-        assert route_definition.router is True
-        assert flow.state.messages[-1].content == "worked"
-
-    def test_legacy_decorated_conversation_start_runs_once_per_turn(
-        self,
-    ) -> None:
-        """Legacy ``@start`` overrides are not invoked again by the router."""
-
-        bootstrap_calls: list[str] = []
-
-        @ConversationConfig()
-        class BootstrapFlow(ConversationalFlow):
-            @start()
-            def conversation_start(self) -> str | None:
-                bootstrap_calls.append("ran")
-                return super().conversation_start()
-
-            def route_turn(self, context: dict[str, Any]) -> str | None:
-                return "work"
-
-            @listen("work")
-            def do_work(self) -> str:
-                self.append_assistant_message("worked")
-                return "worked"
-
-        flow = BootstrapFlow()
-        flow.handle_turn("hi")
-
         assert bootstrap_calls == ["ran"]
         assert flow.state.messages[-1].content == "worked"
 
@@ -1201,40 +1170,6 @@ class TestConversationalFlow:
                     "finalize_session_traces must finalize the trace batch once"
                 )
 
-    def test_deferred_resume_skips_per_resume_flow_finished_event(self) -> None:
-        """Deferred sessions do not emit terminal events while resuming."""
-        from crewai.events.types.flow_events import FlowFinishedEvent
-        from crewai.flow.async_feedback.types import PendingFeedbackContext
-
-        class DeferredResumeFlow(Flow[ChatState]):
-            defer_trace_finalization = True
-
-            @start()
-            def begin(self) -> str:
-                return "started"
-
-        flow = DeferredResumeFlow()
-        flow._pending_feedback_context = PendingFeedbackContext(
-            flow_id=flow.flow_id,
-            flow_class="DeferredResumeFlow",
-            method_name="begin",
-            method_output="started",
-            message="Review",
-        )
-
-        finished_events: list[FlowFinishedEvent] = []
-
-        with crewai_event_bus.scoped_handlers():
-
-            @crewai_event_bus.on(FlowFinishedEvent)
-            def capture(_: Any, event: FlowFinishedEvent) -> None:
-                finished_events.append(event)
-
-            flow.resume("approved")
-            crewai_event_bus.flush()
-
-        assert finished_events == []
-
     def test_finalize_session_traces_restores_event_scope(self, capsys) -> None:
         """No ``empty scope stack`` warning when deferred ``flow_finished`` fires.
 
@@ -1536,44 +1471,6 @@ class TestDeferredFlowLifecycleEvents:
                     listener.batch_manager.finalize_batch()
             mock_finalize.assert_not_called()
 
-    def test_deferred_flow_kickoff_marks_trace_manager_session_deferred(
-        self,
-    ) -> None:
-        class DeferredTraceFlow(Flow[ChatState]):
-            @start()
-            def begin(self) -> str:
-                return "done"
-
-        listener = TraceCollectionListener()
-        listener.batch_manager.defer_session_finalization = False
-
-        flow = DeferredTraceFlow()
-        flow.defer_trace_finalization = True
-
-        with patch.object(listener.batch_manager, "finalize_batch"):
-            flow.kickoff()
-
-        assert listener.batch_manager.defer_session_finalization is True
-
-        flow.finalize_session_traces()
-
-        assert listener.batch_manager.defer_session_finalization is False
-
-    def test_non_deferred_flow_kickoff_clears_stale_trace_manager_flag(
-        self,
-    ) -> None:
-        class PlainTraceFlow(Flow[ChatState]):
-            @start()
-            def begin(self) -> str:
-                return "done"
-
-        listener = TraceCollectionListener()
-        listener.batch_manager.defer_session_finalization = True
-
-        PlainTraceFlow().kickoff()
-
-        assert listener.batch_manager.defer_session_finalization is False
-
 
 class TestNestedCrewTracing:
     def test_is_inside_active_flow_context_when_kickoff_running(self) -> None:
@@ -1627,130 +1524,3 @@ class TestNestedCrewTracing:
             elif listener.batch_manager.batch_owner_type == "crew":
                 listener.batch_manager.finalize_batch()
             mock_finalize.assert_not_called()
-
-    def test_lazy_flow_batch_from_context_preserves_deferred_parent(self) -> None:
-        from crewai.events.listeners.tracing.trace_listener import (
-            TraceCollectionListener,
-        )
-
-        listener = TraceCollectionListener()
-        listener.batch_manager.current_batch = None
-        listener.batch_manager.batch_owner_type = None
-        listener.batch_manager.batch_owner_id = None
-        listener.batch_manager.defer_session_finalization = False
-        listener.batch_manager.event_buffer.clear()
-
-        flow_id_token = current_flow_id.set("parent-flow-id")
-        flow_name_token = current_flow_name.set("ParentChatFlow")
-        defer_token = current_flow_defer_trace_finalization.set(True)
-        try:
-            initialized = listener._try_initialize_flow_batch_from_context(
-                type("Event", (), {"timestamp": None})()
-            )
-
-            assert initialized is True
-            assert listener.batch_manager.batch_owner_type == "flow"
-            assert listener.batch_manager.batch_owner_id == "parent-flow-id"
-            assert listener.batch_manager.defer_session_finalization is True
-            assert listener.batch_manager.current_batch is not None
-            assert (
-                listener.batch_manager.current_batch.execution_metadata[
-                    "execution_type"
-                ]
-                == "flow"
-            )
-            assert (
-                listener.batch_manager.current_batch.execution_metadata["flow_name"]
-                == "ParentChatFlow"
-            )
-        finally:
-            current_flow_defer_trace_finalization.reset(defer_token)
-            current_flow_name.reset(flow_name_token)
-            current_flow_id.reset(flow_id_token)
-            listener.batch_manager.current_batch = None
-            listener.batch_manager.batch_owner_type = None
-            listener.batch_manager.batch_owner_id = None
-            listener.batch_manager.trace_batch_id = None
-            listener.batch_manager.defer_session_finalization = False
-            listener.batch_manager.event_buffer.clear()
-
-    def test_nested_agent_executor_flow_does_not_finalize_parent_batch(
-        self,
-    ) -> None:
-        from crewai import Agent, Crew, Task
-        from crewai.llms.base_llm import BaseLLM
-
-        class StaticLLM(BaseLLM):
-            def __init__(self) -> None:
-                super().__init__(model="debug-static-llm", provider="debug")
-
-            def call(
-                self,
-                messages: Any,
-                tools: Any = None,
-                callbacks: Any = None,
-                available_functions: Any = None,
-                from_task: Any = None,
-                from_agent: Any = None,
-                response_model: Any = None,
-            ) -> str:
-                return (
-                    "Thought: I can answer directly.\n"
-                    "Final Answer: nested crew result"
-                )
-
-        class NestedCrewFlow(Flow[ChatState]):
-            defer_trace_finalization = True
-            tracing = True
-
-            @start()
-            def begin(self) -> str:
-                return "run_nested_crew"
-
-            @listen(begin)
-            def run_nested_crew(self, _: str) -> str:
-                agent = Agent(
-                    role="Debug Agent",
-                    goal="Return a short deterministic result",
-                    backstory="Used only for trace finalization debugging.",
-                    llm=StaticLLM(),
-                    verbose=False,
-                )
-                task = Task(
-                    description="Return the deterministic nested crew result.",
-                    expected_output="nested crew result",
-                    agent=agent,
-                )
-                return Crew(agents=[agent], tasks=[task], verbose=False).kickoff().raw
-
-        listener = TraceCollectionListener()
-        listener.batch_manager.current_batch = None
-        listener.batch_manager.batch_owner_type = None
-        listener.batch_manager.batch_owner_id = None
-        listener.batch_manager.trace_batch_id = None
-        listener.batch_manager.defer_session_finalization = False
-        listener.batch_manager.event_buffer.clear()
-        listener.first_time_handler.is_first_time = False
-
-        def initialize_backend_batch(*_: Any, **__: Any) -> None:
-            listener.batch_manager.trace_batch_id = "debug-trace-batch"
-
-        flow = NestedCrewFlow()
-
-        with (
-            patch.object(
-                listener.batch_manager,
-                "_initialize_backend_batch",
-                side_effect=initialize_backend_batch,
-            ),
-            patch.object(listener.batch_manager, "finalize_batch") as mock_finalize,
-        ):
-            flow.kickoff()
-            crewai_event_bus.flush()
-            flow.kickoff()
-            crewai_event_bus.flush()
-
-            assert mock_finalize.call_count == 0, (
-                "nested AgentExecutor flows inside a deferred parent Flow must "
-                "not finalize the parent trace batch"
-            )

lib/crewai/tests/test_flow_definition.py

@@ -223,11 +223,10 @@ def test_flow_definition_includes_conversational_builtins_when_enabled():
     assert definition.conversational.enabled is True
     assert definition.conversational.defer_trace_finalization is True
     assert definition.conversational.builtin_routes == ["converse", "end"]
-    assert "conversation_start" not in methods
+    assert "conversation_start" in methods
     assert "route_conversation" in methods
     assert "converse_turn" in methods
-    assert methods["route_conversation"].start is True
-    assert methods["route_conversation"].router is True
+    assert methods["conversation_start"].start is True
 
 
 def test_flow_definition_serializes_conversational_config():
@@ -261,7 +260,7 @@ def test_flow_definition_serializes_conversational_config():
     assert conversational.router.fallback_intent == "end"
 
 
-def test_flow_definition_uses_collapsed_conversational_router_start():
+def test_flow_definition_preserves_undecorated_conversational_override():
     class ChatFlow(Flow):
         conversational = True
 
@@ -270,10 +269,8 @@ def test_flow_definition_uses_collapsed_conversational_router_start():
 
     methods = ChatFlow.flow_definition().methods
 
-    assert "conversation_start" not in methods
+    assert methods["conversation_start"].start is True
     assert "route_conversation" in methods
-    assert methods["route_conversation"].start is True
-    assert methods["route_conversation"].router is True
 
 
 def test_flow_definition_serializes_human_feedback_metadata():

lib/crewai/tests/test_flow_from_definition.py

@@ -0,0 +1,508 @@
+from __future__ import annotations
+
+import pytest
+
+from crewai.events.event_bus import crewai_event_bus
+from crewai.events.types.flow_events import (
+    MethodExecutionFinishedEvent,
+    MethodExecutionStartedEvent,
+)
+from crewai.flow import Flow, and_, listen, or_, router, start
+from crewai.flow.flow import FlowState
+from crewai.flow.flow_definition import FlowDefinition
+
+
+class ChainFlow(Flow):
+    @start()
+    def begin(self):
+        self.state["begin_ran"] = True
+        return "hello"
+
+    @listen(begin)
+    def shout(self, result):
+        return result.upper()
+
+    @listen(shout)
+    def confirm(self):
+        self.state["confirmed"] = True
+        return f"confirmed:{self.state['confirmed']}"
+
+
+CHAIN_YAML = f"""
+schema: crewai.flow/v1
+name: ChainFlow
+methods:
+  begin:
+    handler: {__name__}:ChainFlow.begin
+    start: true
+  shout:
+    handler: {__name__}:ChainFlow.shout
+    listen: begin
+  confirm:
+    handler: {__name__}:ChainFlow.confirm
+    listen: shout
+"""
+
+
+class MergeFlow(Flow):
+    @start()
+    def begin(self):
+        return "go"
+
+    @listen(begin)
+    def left(self):
+        return "left"
+
+    @listen(begin)
+    def right(self):
+        return "right"
+
+    @listen(or_(left, right))
+    def either(self):
+        self.state["either_ran"] = True
+        return "either"
+
+    @listen(and_(left, right, either))
+    def join(self):
+        self.state["joined"] = True
+        return "joined"
+
+
+MERGE_YAML = f"""
+schema: crewai.flow/v1
+name: MergeFlow
+methods:
+  begin:
+    handler: {__name__}:MergeFlow.begin
+    start: true
+  left:
+    handler: {__name__}:MergeFlow.left
+    listen: begin
+  right:
+    handler: {__name__}:MergeFlow.right
+    listen: begin
+  either:
+    handler: {__name__}:MergeFlow.either
+    listen:
+      or: [left, right]
+  join:
+    handler: {__name__}:MergeFlow.join
+    listen:
+      and: [left, right, either]
+"""
+
+
+class RouteFlow(Flow):
+    @start()
+    def begin(self):
+        return "go"
+
+    @router(begin)
+    def decide(self):
+        return "left" if self.state.get("direction") == "left" else "right"
+
+    @listen("left")
+    def take_left(self):
+        return "took-left"
+
+    @listen("right")
+    def take_right(self):
+        return "took-right"
+
+
+ROUTE_YAML = f"""
+schema: crewai.flow/v1
+name: RouteFlow
+methods:
+  begin:
+    handler: {__name__}:RouteFlow.begin
+    start: true
+  decide:
+    handler: {__name__}:RouteFlow.decide
+    listen: begin
+    router: true
+  take_left:
+    handler: {__name__}:RouteFlow.take_left
+    listen: left
+  take_right:
+    handler: {__name__}:RouteFlow.take_right
+    listen: right
+"""
+
+
+class LoopFlow(Flow):
+    @start("retry")
+    def step(self):
+        self.state["count"] = self.state.get("count", 0) + 1
+        return self.state["count"]
+
+    @router(step)
+    def decide(self):
+        if self.state["count"] < 3:
+            return "retry"
+        return "done"
+
+    @listen("done")
+    def finish(self):
+        return f"finished:{self.state['count']}"
+
+
+LOOP_YAML = f"""
+schema: crewai.flow/v1
+name: LoopFlow
+methods:
+  step:
+    handler: {__name__}:LoopFlow.step
+    start: retry
+  decide:
+    handler: {__name__}:LoopFlow.decide
+    listen: step
+    router: true
+  finish:
+    handler: {__name__}:LoopFlow.finish
+    listen: done
+"""
+
+
+class CounterState(FlowState):
+    count: int = 0
+    label: str = "none"
+
+
+class PydanticStateFlow(Flow[CounterState]):
+    @start()
+    def begin(self):
+        self.state.count += 1
+        return self.state.count
+
+    @listen(begin)
+    def finish(self):
+        self.state.label = f"count={self.state.count}"
+        return self.state.label
+
+
+PYDANTIC_STATE_YAML = f"""
+schema: crewai.flow/v1
+name: PydanticStateFlow
+state:
+  type: pydantic
+  ref: {__name__}:CounterState
+methods:
+  begin:
+    handler: {__name__}:PydanticStateFlow.begin
+    start: true
+  finish:
+    handler: {__name__}:PydanticStateFlow.finish
+    listen: begin
+"""
+
+PYDANTIC_STATE_OVERLAY_YAML = f"""
+schema: crewai.flow/v1
+name: PydanticStateFlow
+state:
+  type: pydantic
+  ref: {__name__}:CounterState
+  default:
+    count: 5
+methods:
+  begin:
+    handler: {__name__}:PydanticStateFlow.begin
+    start: true
+  finish:
+    handler: {__name__}:PydanticStateFlow.finish
+    listen: begin
+"""
+
+JSON_SCHEMA_STATE_YAML = f"""
+schema: crewai.flow/v1
+name: JsonSchemaStateFlow
+state:
+  type: json_schema
+  json_schema:
+    title: CounterState
+    type: object
+    properties:
+      count:
+        type: integer
+        default: 0
+      label:
+        type: string
+        default: none
+methods:
+  begin:
+    handler: {__name__}:PydanticStateFlow.begin
+    start: true
+  finish:
+    handler: {__name__}:PydanticStateFlow.finish
+    listen: begin
+"""
+
+PYDANTIC_REF_WITH_SCHEMA_FALLBACK_YAML = f"""
+schema: crewai.flow/v1
+name: SchemaFallbackFlow
+state:
+  type: pydantic
+  ref: definitely_not_a_module_xyz:MissingState
+  json_schema:
+    title: CounterState
+    type: object
+    properties:
+      count:
+        type: integer
+        default: 0
+      label:
+        type: string
+        default: none
+methods:
+  begin:
+    handler: {__name__}:PydanticStateFlow.begin
+    start: true
+  finish:
+    handler: {__name__}:PydanticStateFlow.finish
+    listen: begin
+"""
+
+UNRESOLVABLE_STATE_YAML = f"""
+schema: crewai.flow/v1
+name: UnresolvableStateFlow
+state:
+  type: pydantic
+  ref: definitely_not_a_module_xyz:MissingState
+methods:
+  begin:
+    handler: {__name__}:ChainFlow.begin
+    start: true
+"""
+
+DICT_STATE_YAML = f"""
+schema: crewai.flow/v1
+name: DictStateFlow
+state:
+  type: dict
+  default:
+    count: 5
+methods:
+  begin:
+    handler: {__name__}:ChainFlow.begin
+    start: true
+"""
+
+UNKNOWN_STATE_YAML = f"""
+schema: crewai.flow/v1
+name: UnknownStateFlow
+state:
+  type: unknown
+  ref: somewhere:Something
+methods:
+  begin:
+    handler: {__name__}:ChainFlow.begin
+    start: true
+"""
+
+
+def _run_with_events(flow, inputs=None):
+    events = []
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(MethodExecutionStartedEvent)
+        def on_started(source, event):
+            events.append(event)
+
+        @crewai_event_bus.on(MethodExecutionFinishedEvent)
+        def on_finished(source, event):
+            events.append(event)
+
+        result = flow.kickoff(inputs=inputs)
+    events.sort(key=lambda e: e.timestamp)
+    return result, [
+        (type(e).__name__, str(e.method_name), e.flow_name) for e in events
+    ]
+
+
+def _state_without_id(flow):
+    snapshot = dict(flow.state.model_dump())
+    snapshot.pop("id", None)
+    return snapshot
+
+
+def assert_parity(flow_cls, yaml_str, inputs=None, ordered=True):
+    class_flow = flow_cls()
+    class_result, class_events = _run_with_events(class_flow, inputs)
+
+    definition = FlowDefinition.from_yaml(yaml_str)
+    definition_flow = Flow.from_definition(definition)
+    definition_result, definition_events = _run_with_events(definition_flow, inputs)
+
+    assert definition_result == class_result
+    assert _state_without_id(definition_flow) == _state_without_id(class_flow)
+    if ordered:
+        assert definition_flow.method_outputs == class_flow.method_outputs
+        assert definition_events == class_events
+    else:
+        assert sorted(map(repr, definition_flow.method_outputs)) == sorted(
+            map(repr, class_flow.method_outputs)
+        )
+        assert sorted(definition_events) == sorted(class_events)
+    return definition_flow, definition_result
+
+
+def test_simple_chain_parity():
+    flow, result = assert_parity(ChainFlow, CHAIN_YAML)
+    assert result == "confirmed:True"
+    assert flow.method_outputs == ["hello", "HELLO", "confirmed:True"]
+
+
+def test_and_or_merge_parity():
+    flow, _ = assert_parity(MergeFlow, MERGE_YAML, ordered=False)
+    assert flow.state["joined"] is True
+    assert flow.state["either_ran"] is True
+
+
+def test_router_label_parity_for_each_branch():
+    left_flow, _ = assert_parity(RouteFlow, ROUTE_YAML, inputs={"direction": "left"})
+    assert "took-left" in left_flow.method_outputs
+    assert "took-right" not in left_flow.method_outputs
+
+    right_flow, _ = assert_parity(RouteFlow, ROUTE_YAML, inputs={"direction": "right"})
+    assert "took-right" in right_flow.method_outputs
+
+
+def test_cyclic_flow_parity():
+    flow, result = assert_parity(LoopFlow, LOOP_YAML)
+    assert result == "finished:3"
+    assert flow.state["count"] == 3
+
+
+def test_definition_flow_events_use_definition_name():
+    definition = FlowDefinition.from_yaml(CHAIN_YAML)
+    flow = Flow.from_definition(definition)
+    _, events = _run_with_events(flow)
+    assert events
+    assert all(flow_name == "ChainFlow" for _, _, flow_name in events)
+
+
+def test_from_definition_missing_handler_raises():
+    definition = FlowDefinition.from_dict(
+        {
+            "schema": "crewai.flow/v1",
+            "name": "NoHandlers",
+            "methods": {"begin": {"start": True}},
+        }
+    )
+
+    with pytest.raises(ValueError, match="begin: no handler"):
+        Flow.from_definition(definition)
+
+
+def test_from_definition_unresolvable_handler_raises():
+    definition = FlowDefinition.from_dict(
+        {
+            "schema": "crewai.flow/v1",
+            "name": "BadHandlers",
+            "methods": {
+                "begin": {
+                    "start": True,
+                    "handler": "definitely_not_a_module_xyz:nope",
+                }
+            },
+        }
+    )
+
+    with pytest.raises(ValueError, match="missing or unresolvable handlers.*begin"):
+        Flow.from_definition(definition)
+
+
+def test_from_definition_malformed_handler_raises():
+    definition = FlowDefinition.from_dict(
+        {
+            "schema": "crewai.flow/v1",
+            "name": "MalformedHandlers",
+            "methods": {"begin": {"start": True, "handler": "no-colon-here"}},
+        }
+    )
+
+    with pytest.raises(ValueError, match="expected 'module:qualname'"):
+        Flow.from_definition(definition)
+
+
+def test_flow_definition_stamps_handler_refs():
+    definition = ChainFlow.flow_definition()
+
+    assert definition.methods["begin"].handler == f"{__name__}:ChainFlow.begin"
+    assert definition.methods["shout"].handler == f"{__name__}:ChainFlow.shout"
+
+
+def test_pydantic_state_from_ref_parity():
+    flow, result = assert_parity(PydanticStateFlow, PYDANTIC_STATE_YAML)
+    assert result == "count=1"
+    assert flow.state.count == 1
+    assert flow.state.label == "count=1"
+    assert flow.state.id
+
+
+def test_pydantic_state_default_overlay():
+    flow = Flow.from_definition(FlowDefinition.from_yaml(PYDANTIC_STATE_OVERLAY_YAML))
+    result = flow.kickoff()
+    assert result == "count=6"
+    assert flow.state.count == 6
+
+
+def test_json_schema_state():
+    flow = Flow.from_definition(FlowDefinition.from_yaml(JSON_SCHEMA_STATE_YAML))
+    result = flow.kickoff()
+    assert result == "count=1"
+    assert flow.state.count == 1
+    assert flow.state.label == "count=1"
+    assert flow.state.id
+
+
+def test_json_schema_state_validates_inputs():
+    flow = Flow.from_definition(FlowDefinition.from_yaml(JSON_SCHEMA_STATE_YAML))
+    with pytest.raises(ValueError, match="Invalid inputs"):
+        flow.kickoff(inputs={"count": "not-a-number"})
+
+
+def test_pydantic_state_falls_back_to_json_schema_when_ref_unimportable():
+    flow = Flow.from_definition(
+        FlowDefinition.from_yaml(PYDANTIC_REF_WITH_SCHEMA_FALLBACK_YAML)
+    )
+    result = flow.kickoff()
+    assert result == "count=1"
+    assert flow.state.count == 1
+
+
+def test_pydantic_state_without_ref_or_schema_falls_back_to_dict(caplog):
+    with caplog.at_level("ERROR"):
+        flow = Flow.from_definition(FlowDefinition.from_yaml(UNRESOLVABLE_STATE_YAML))
+    assert "falling back to dict state" in caplog.text
+
+    result = flow.kickoff()
+    assert result == "hello"
+    assert flow.state["begin_ran"] is True
+    assert flow.state["id"]
+
+
+def test_dict_state_is_a_copy_of_default_plus_id():
+    definition = FlowDefinition.from_yaml(DICT_STATE_YAML)
+
+    flow = Flow.from_definition(definition)
+    assert flow.state["count"] == 5
+    assert flow.state["id"]
+    flow.kickoff()
+    assert flow.state["begin_ran"] is True
+
+    second = Flow.from_definition(definition)
+    assert second.state["count"] == 5
+    assert "begin_ran" not in second.state.model_dump()
+    assert second.state["id"] != flow.state["id"]
+    assert definition.state.default == {"count": 5}
+
+
+def test_unknown_state_type_falls_back_to_dict(caplog):
+    with caplog.at_level("WARNING"):
+        flow = Flow.from_definition(FlowDefinition.from_yaml(UNKNOWN_STATE_YAML))
+    assert "falling back to dict state" in caplog.text
+
+    result = flow.kickoff()
+    assert result == "hello"
+    assert flow.state["begin_ran"] is True

lib/devtools/src/crewai_devtools/__init__.py

@@ -1,3 +1,3 @@
 """CrewAI development tools."""
 
-__version__ = "1.14.7rc1"
+__version__ = "1.14.7a4"