crewAI/lib/crewai/tests/test_flow_definition.py

"""Tests for the static Flow Definition contract."""

from enum import Enum
import importlib
import inspect
import logging
from pathlib import Path
import re
from typing import Annotated, Literal

import pytest
from pydantic import BaseModel, ValidationError

import crewai.flow.dsl as flow_dsl
import crewai.flow.flow_definition as flow_definition
import crewai.flow.visualization.builder as visualization_builder
from crewai.experimental import ConversationConfig, RouterConfig
from crewai.flow.expressions import (
    FLOW_TEMPLATE_EXPRESSION_EXAMPLES,
    FLOW_TEMPLATE_EXPRESSION_RULES,
)
from crewai.flow import Flow, and_, human_feedback, listen, or_, persist, router, start


def test_flow_public_exports_are_explicit():
    import crewai.flow.visualization as flow_visualization

    flow_package = importlib.import_module("crewai.flow")

    assert "FlowDefinition" not in flow_package.__all__
    assert "FlowDefinitionDiagnostic" not in flow_package.__all__
    assert "build_flow_definition" not in flow_package.__all__
    assert "flow_structure" not in flow_package.__all__
    assert set(flow_dsl.__all__) == {
        "HumanFeedbackResult",
        "and_",
        "human_feedback",
        "listen",
        "or_",
        "router",
        "start",
    }
    assert set(flow_definition.__all__) == {
        "FlowActionDefinition",
        "FlowAgentActionDefinition",
        "FlowAtomicActionDefinition",
        "FlowCodeActionDefinition",
        "FlowConfigDefinition",
        "FlowConversationalDefinition",
        "FlowConversationalRouterDefinition",
        "FlowCrewActionDefinition",
        "FlowDefinition",
        "FlowDefinitionCondition",
        "FlowDictStateDefinition",
        "FlowEachActionDefinition",
        "FlowEachStepDefinition",
        "FlowExpressionActionDefinition",
        "FlowHumanFeedbackDefinition",
        "FlowJsonSchemaStateDefinition",
        "FlowMethodDefinition",
        "FlowPersistenceDefinition",
        "FlowPydanticStateDefinition",
        "FlowScriptActionDefinition",
        "FlowStateDefinition",
        "FlowToolActionDefinition",
        "FlowUnknownStateDefinition",
    }
    assert "build_flow_structure" in flow_visualization.__all__
    assert "calculate_node_levels" not in flow_visualization.__all__


def test_flow_definition_json_schema_carries_reference_descriptions():
    schema = flow_definition.FlowDefinition.model_json_schema(by_alias=True)
    defs = schema["$defs"]

    assert schema["properties"]["schema"]["description"]
    assert schema["properties"]["methods"]["description"]
    assert "diagnostics" not in schema["properties"]

    method_properties = defs["FlowMethodDefinition"]["properties"]
    assert method_properties["do"]["description"] == "Action executed when this method runs."
    assert "Trigger condition" in method_properties["listen"]["description"]

    script_properties = defs["FlowScriptActionDefinition"]["properties"]
    assert "trusted inline Python" in script_properties["call"]["description"]
    assert "not interpolated" in script_properties["code"]["description"]
    assert "not sandboxed" in script_properties["code"]["description"]

    agent_properties = defs["FlowAgentActionDefinition"]["properties"]
    assert "Individual Agent definition" in agent_properties["with"]["description"]
    assert "outside of a crew" in agent_properties["with"]["description"]
    assert "individual inline Agent" in agent_properties["call"]["description"]

    expression_rule = FLOW_TEMPLATE_EXPRESSION_RULES[0]
    code_properties = defs["FlowCodeActionDefinition"]["properties"]
    tool_properties = defs["FlowToolActionDefinition"]["properties"]
    crew_properties = defs["FlowCrewActionDefinition"]["properties"]
    assert expression_rule in code_properties["with"]["description"]
    assert expression_rule in tool_properties["with"]["description"]
    assert expression_rule in crew_properties["inputs"]["description"]

    state_schema = next(
        branch
        for branch in schema["properties"]["state"]["anyOf"]
        if "discriminator" in branch
    )
    assert state_schema["discriminator"]["propertyName"] == "type"
    assert state_schema["discriminator"]["mapping"] == {
        "dict": "#/$defs/FlowDictStateDefinition",
        "json_schema": "#/$defs/FlowJsonSchemaStateDefinition",
        "pydantic": "#/$defs/FlowPydanticStateDefinition",
        "unknown": "#/$defs/FlowUnknownStateDefinition",
    }

    dict_state_properties = defs["FlowDictStateDefinition"]["properties"]
    assert dict_state_properties["type"]["description"]
    assert "ref" not in dict_state_properties

    json_schema_state_properties = defs["FlowJsonSchemaStateDefinition"]["properties"]
    assert json_schema_state_properties["json_schema"]["description"]
    assert "json_schema" in defs["FlowJsonSchemaStateDefinition"]["required"]

    pydantic_state_properties = defs["FlowPydanticStateDefinition"]["properties"]
    assert "Fallback JSON Schema" in pydantic_state_properties["json_schema"][
        "description"
    ]

    each_properties = defs["FlowEachActionDefinition"]["properties"]
    assert "list to iterate" in each_properties["in"]["description"]
    assert "Ordered steps" in each_properties["do"]["description"]

    step_properties = defs["FlowEachStepDefinition"]["properties"]
    assert "runs only if" in step_properties["if"]["description"]


def test_flow_definition_json_schema_carries_field_examples_only():
    schema = flow_definition.FlowDefinition.model_json_schema(by_alias=True)
    defs = schema["$defs"]

    for model_name in [
        "FlowDefinition",
        "FlowCodeActionDefinition",
        "FlowToolActionDefinition",
        "FlowAgentActionDefinition",
        "FlowCrewActionDefinition",
        "FlowExpressionActionDefinition",
        "FlowScriptActionDefinition",
        "FlowEachActionDefinition",
        "FlowEachStepDefinition",
        "FlowMethodDefinition",
        "FlowDictStateDefinition",
        "FlowJsonSchemaStateDefinition",
        "FlowPydanticStateDefinition",
        "FlowUnknownStateDefinition",
        "FlowConfigDefinition",
        "FlowPersistenceDefinition",
        "FlowHumanFeedbackDefinition",
    ]:
        model_schema = schema if model_name == "FlowDefinition" else defs[model_name]
        assert "examples" not in model_schema

    assert schema["properties"]["name"]["examples"] == ["ResearchFlow"]
    assert schema["properties"]["schema"]["examples"] == ["crewai.flow/v1"]
    assert schema["properties"]["methods"]["examples"][0]["seed"]["do"] == {
        "call": "expression",
        "expr": "state.topic",
    }

    script_properties = defs["FlowScriptActionDefinition"]["properties"]
    assert script_properties["call"]["examples"] == ["script"]
    assert "state['topic'].strip()" in script_properties["code"]["examples"][0]
    assert script_properties["language"]["examples"] == ["python"]

    action_properties = defs["FlowCodeActionDefinition"]["properties"]
    assert action_properties["ref"]["examples"] == [
        "my_project.flows:normalize_topic"
    ]
    assert action_properties["with"]["examples"] == [
        {"topic": "${state.topic}", "query": "News about ${state.topic}"}
    ]

    agent_properties = defs["FlowAgentActionDefinition"]["properties"]
    assert agent_properties["call"]["examples"] == ["agent"]
    assert agent_properties["with"]["examples"][0]["input"] == "${state.question}"

    each_properties = defs["FlowEachActionDefinition"]["properties"]
    assert each_properties["in"]["examples"] == ["state.rows"]
    assert each_properties["do"]["examples"][0][0]["name"] == "clean"
    assert each_properties["do"]["examples"][0][0]["action"]["call"] == "script"
    assert each_properties["do"]["examples"][0][1]["if"] == "outputs.clean != ''"

    step_properties = defs["FlowEachStepDefinition"]["properties"]
    assert step_properties["if"]["examples"] == ["item.kind == 'invoice'"]

    method_properties = defs["FlowMethodDefinition"]["properties"]
    assert method_properties["listen"]["examples"] == [
        "seed",
        {"or": ["approved", "revise"]},
    ]
    assert method_properties["emit"]["examples"] == [["approved", "revise"]]


def test_flow_state_definition_uses_discriminated_branches():
    definition = flow_definition.FlowDefinition.model_validate(
        {
            "name": "TypedStateFlow",
            "state": {
                "type": "json_schema",
                "json_schema": {"type": "object"},
            },
        }
    )

    assert isinstance(
        definition.state,
        flow_definition.FlowJsonSchemaStateDefinition,
    )

    with pytest.raises(ValidationError, match="extra_forbidden"):
        flow_definition.FlowDefinition.model_validate(
            {
                "name": "InvalidStateFlow",
                "state": {
                    "type": "dict",
                    "ref": "my_project.flows:ResearchState",
                },
            }
        )


def test_condition_combinators_return_nested_runtime_tree():
    condition = and_("event_a", "event_b", or_("event_c"))

    assert condition == {
        "type": "AND",
        "conditions": [
            "event_a",
            "event_b",
            {"type": "OR", "conditions": ["event_c"]},
        ],
    }


def test_flow_definition_lowers_nested_conditions():
    class NestedFlow(Flow):
        @start()
        def begin(self):
            return "begin"

        @listen(begin)
        def validated(self):
            return "validated"

        @listen(begin)
        def processed(self):
            return "processed"

        @listen(or_(and_(validated, processed), begin))
        def finalize(self):
            return "done"

    finalize = NestedFlow.flow_definition().methods["finalize"]

    assert finalize.listen == {"or": [{"and": ["validated", "processed"]}, "begin"]}


def test_flow_definition_preserves_single_branch_nested_conditions():
    class AmbiguousFlow(Flow):
        @start()
        def event_a(self):
            return "a"

        @listen(event_a)
        def event_b(self):
            return "b"

        @listen(and_(event_a, event_b, or_("event_c")))
        def event_d(self):
            return "d"

    event_d = AmbiguousFlow.flow_definition().methods["event_d"]

    assert event_d.listen == {"and": ["event_a", "event_b", {"or": ["event_c"]}]}


def test_flow_definition_rejects_invalid_condition():
    with pytest.raises(ValueError, match="Invalid condition"):
        start(123)(lambda self: None)


def test_flow_definition_contract_is_dsl_agnostic():
    source_path = Path(inspect.getsourcefile(flow_definition) or "")
    source = source_path.read_text()

    assert "DSL" not in source
    assert "flow_wrappers" not in source
    assert "build_flow_definition" not in source
    assert "extract_flow_definition" not in source


def test_flow_definition_maps_dsl_to_static_contract():
    class ContractState(BaseModel):
        topic: str = ""

    class ContractFlow(Flow[ContractState]):
        """A flow with every core DSL role."""

        initial_state = ContractState
        stream = True
        max_method_calls = 7

        @start()
        def begin(self):
            return "started"

        @listen(begin)
        def process(self):
            return "processed"

        @router(process)
        def decide(self):
            return "approved"

        @listen(or_("approved", "revise"))
        @human_feedback(
            message="Review this output.",
            emit=["done", "revise"],
            llm="gpt-4o-mini",
            default_outcome="done",
            metadata={"team": "qa"},
            learn=True,
            learn_source="hitl",
            learn_strict=True,
        )
        def review(self):
            return "review"

        @listen(and_(begin, process))
        def audit(self):
            return "audit"

    definition = ContractFlow.flow_definition()

    assert definition.schema_ == "crewai.flow/v1"
    assert definition.name == "ContractFlow"
    assert definition.description == "A flow with every core DSL role."
    assert definition.state is not None
    assert definition.state.type == "pydantic"
    assert definition.state.ref and "ContractState" in definition.state.ref
    assert definition.config.stream is True
    assert definition.config.max_method_calls == 7
    assert definition.conversational is None

    assert definition.methods["begin"].start is True
    assert definition.methods["process"].listen == "begin"

    decide = definition.methods["decide"]
    assert decide.listen == "process"
    assert decide.router is True
    assert decide.emit is None

    review = definition.methods["review"]
    assert review.listen == {"or": ["approved", "revise"]}
    assert review.router is True
    assert review.emit is None
    assert review.human_feedback is not None
    assert review.human_feedback.emit == ["done", "revise"]
    assert review.human_feedback.default_outcome == "done"
    assert review.human_feedback.metadata == {"team": "qa"}
    assert review.human_feedback.learn is True
    assert review.human_feedback.learn_strict is True

    assert definition.methods["audit"].listen == {"and": ["begin", "process"]}
    assert "diagnostics" not in definition.to_dict()


def test_flow_definition_excludes_conversational_builtins_for_regular_flows():
    class RegularFlow(Flow):
        @start()
        def begin(self):
            return "begin"

    methods = RegularFlow.flow_definition().methods

    assert RegularFlow.flow_definition().conversational is None
    assert set(methods) == {"begin"}
    assert "conversation_start" not in methods
    assert "route_conversation" not in methods
    assert "converse_turn" not in methods


def test_flow_definition_includes_conversational_builtins_when_enabled():
    class ChatFlow(Flow):
        conversational = True

    definition = ChatFlow.flow_definition()
    methods = definition.methods

    assert definition.conversational is not None
    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 "route_conversation" in methods
    assert "converse_turn" in methods
    assert methods["route_conversation"].start is True
    assert methods["route_conversation"].router is True


def test_flow_definition_serializes_conversational_config():
    @ConversationConfig(
        system_prompt="Be concise.",
        llm="gpt-4o-mini",
        router=RouterConfig(
            prompt="Pick a route.",
            routes=["research"],
            default_intent="converse",
            fallback_intent="end",
        ),
        default_intents=["research"],
        visible_agent_outputs=["researcher"],
        defer_trace_finalization=False,
    )
    class ChatFlow(Flow):
        conversational = True

    conversational = ChatFlow.flow_definition().conversational

    assert conversational is not None
    assert conversational.system_prompt == "Be concise."
    assert conversational.llm == "gpt-4o-mini"
    assert conversational.default_intents == ["research"]
    assert conversational.visible_agent_outputs == ["researcher"]
    assert conversational.defer_trace_finalization is False
    assert conversational.router is not None
    assert conversational.router.prompt == "Pick a route."
    assert conversational.router.routes == ["research"]
    assert conversational.router.fallback_intent == "end"


def test_flow_definition_uses_collapsed_conversational_router_start():
    class ChatFlow(Flow):
        conversational = True

        def conversation_start(self) -> str | None:
            return "custom"

    methods = ChatFlow.flow_definition().methods

    assert "conversation_start" not in methods
    assert "route_conversation" in methods
    assert methods["route_conversation"].start is True
    assert methods["route_conversation"].router is True


def test_declaring_the_conversational_block_opts_in_without_enabled():
    definition = flow_definition.FlowDefinition.from_declaration(
        contents={
            "schema": "crewai.flow/v1",
            "name": "JsonChat",
            "conversational": {"llm": "gpt-4o-mini"},
            "methods": {
                "begin": {"do": {"call": "expression", "expr": "'x'"}, "start": True}
            },
        }
    )

    assert definition.conversational is not None
    assert definition.conversational.enabled is True
    assert definition.conversational.llm == "gpt-4o-mini"


def test_conversational_block_can_be_explicitly_disabled():
    definition = flow_definition.FlowDefinition.from_declaration(
        contents={
            "schema": "crewai.flow/v1",
            "name": "JsonChat",
            "conversational": {"enabled": False, "llm": "gpt-4o-mini"},
            "methods": {
                "begin": {"do": {"call": "expression", "expr": "'x'"}, "start": True}
            },
        }
    )

    assert definition.conversational is not None
    assert definition.conversational.enabled is False
    assert definition.conversational.llm == "gpt-4o-mini"


def test_omitting_the_conversational_block_leaves_it_none():
    definition = flow_definition.FlowDefinition.from_declaration(
        contents={
            "schema": "crewai.flow/v1",
            "name": "PlainJson",
            "methods": {
                "begin": {"do": {"call": "expression", "expr": "'x'"}, "start": True}
            },
        }
    )

    assert definition.conversational is None


def test_flow_definition_includes_conversational_from_decorator_alone():
    @ConversationConfig(llm="gpt-4o-mini")
    class DecoratedFlow(Flow):
        pass

    definition = DecoratedFlow.flow_definition()

    assert definition.conversational is not None
    assert definition.conversational.enabled is True
    assert "route_conversation" in definition.methods
    assert "converse_turn" in definition.methods


def test_flow_definition_degrades_human_feedback_metadata(caplog):
    caplog.set_level(logging.WARNING, logger="crewai.flow.dsl._utils")
    marker = object()

    class MetadataFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @listen(begin)
        @human_feedback(message="Review this output.", metadata={"marker": marker})
        def review(self):
            return "review"

    definition = MetadataFlow.flow_definition()
    review = definition.methods["review"]

    assert review.human_feedback is not None
    assert review.human_feedback.metadata == {"ref": "builtins:dict"}
    assert any(
        "methods.review.human_feedback.metadata" in record.message
        and "not fully serializable" in record.message
        for record in caplog.records
    )
    definition.to_dict()


def test_flow_definition_fragments_cover_start_listen_and_condition_sugar():
    class FragmentFlow(Flow):
        @start()
        def begin(self):
            return "begin"

        @start("restart_event")
        def restart(self):
            return "restart"

        @listen(begin)
        def by_callable(self):
            return "callable"

        @listen("manual_event")
        def by_string(self):
            return "string"

        @listen(and_(begin, by_callable))
        def by_and(self):
            return "and"

        @listen(or_(and_("manual_event", by_string), "fallback_event"))
        def nested(self):
            return "nested"

    definition = FragmentFlow.flow_definition()

    assert definition.methods["begin"].start is True
    assert definition.methods["restart"].start == "restart_event"
    assert definition.methods["by_callable"].listen == "begin"
    assert definition.methods["by_string"].listen == "manual_event"
    assert definition.methods["by_and"].listen == {"and": ["begin", "by_callable"]}
    assert definition.methods["nested"].listen == {
        "or": [{"and": ["manual_event", "by_string"]}, "fallback_event"]
    }

    assert not hasattr(FragmentFlow.__dict__["begin"], "__is_start_method__")
    assert not hasattr(FragmentFlow.__dict__["restart"], "__trigger_methods__")
    for method_name in ("by_callable", "by_string", "by_and", "nested"):
        method = FragmentFlow.__dict__[method_name]
        assert not hasattr(method, "__trigger_methods__")
        assert not hasattr(method, "__condition_type__")
        assert not hasattr(method, "__trigger_condition__")


def test_human_feedback_emit_overrides_inner_router_emit():
    class FeedbackOverRouterFlow(Flow):
        @start()
        def begin(self):
            return "data"

        @human_feedback(
            message="Review:",
            emit=["approved", "rejected"],
            llm="gpt-4o-mini",
        )
        @router(begin, emit=["x", "y"])
        def route(self):
            return "approved"

        @listen("approved")
        def proceed(self):
            return "ok"

    route = FeedbackOverRouterFlow.flow_definition().methods["route"]
    assert route.router is True
    assert route.human_feedback is not None
    assert route.human_feedback.emit == ["approved", "rejected"]
    assert route.emit is None


def test_flow_definition_classifies_start_router_from_human_feedback_emit():
    class StartRouterFlow(Flow):
        @start()
        @human_feedback(
            message="Review:",
            emit=["continue", "stop"],
            llm="gpt-4o-mini",
        )
        def entry_point(self):
            return "data"

        @listen("continue")
        def proceed(self):
            return "proceeding"

        @listen("stop")
        def halt(self):
            return "halted"

    definition = StartRouterFlow.flow_definition()
    entry_point = definition.methods["entry_point"]

    assert entry_point.is_start is True
    assert entry_point.router is True
    assert entry_point.human_feedback is not None
    assert entry_point.human_feedback.emit == ["continue", "stop"]
    assert entry_point.emit is None


def test_flow_definition_classifies_public_dsl_start_router():
    class StartRouterFlow(Flow):
        @start()
        @router(emit=["continue", "stop"])
        def entry_point(self):
            return "continue"

        @router(emit=["resume"])
        @start()
        def alternate_entry_point(self):
            return "resume"

    entry_point = StartRouterFlow.flow_definition().methods["entry_point"]
    alternate_entry_point = StartRouterFlow.flow_definition().methods[
        "alternate_entry_point"
    ]

    assert entry_point.is_start is True
    assert entry_point.router is True
    assert entry_point.listen is None
    assert entry_point.emit == ["continue", "stop"]
    assert alternate_entry_point.is_start is True
    assert alternate_entry_point.router is True
    assert alternate_entry_point.listen is None
    assert alternate_entry_point.emit == ["resume"]


def test_flow_definition_merges_stacked_listen_router():
    class ChainedRouterFlow(Flow):
        @start()
        @router(emit=["approved", "not_approved"])
        def first_router(self):
            return "approved"

        @listen("approved")
        @router(emit=["second_approval", "not_approved"])
        def second_router(self):
            return "second_approval"

    methods = ChainedRouterFlow.flow_definition().methods

    assert methods["first_router"].is_start is True
    assert methods["first_router"].listen is None
    assert methods["second_router"].router is True
    assert methods["second_router"].listen == "approved"
    assert methods["second_router"].emit == ["second_approval", "not_approved"]


def test_flow_definition_from_declaration_accepts_json_and_yaml_strings():
    class RoundTripFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def decide(self):
            return "left"

        @listen("left")
        def handle_left(self):
            return "left"

    expected = RoundTripFlow.flow_definition()
    declarations = [
        """
        {
          "schema": "crewai.flow/v1",
          "name": "RoundTripFlow",
          "methods": {
            "begin": {
              "start": true,
              "do": {
                "call": "code",
                "ref": "test_flow_definition:RoundTripFlow.begin"
              }
            },
            "decide": {
              "listen": "begin",
              "router": true,
              "do": {
                "call": "code",
                "ref": "test_flow_definition:RoundTripFlow.decide"
              }
            },
            "handle_left": {
              "listen": "left",
              "do": {
                "call": "code",
                "ref": "test_flow_definition:RoundTripFlow.handle_left"
              }
            }
          }
        }
        """,
        """
        schema: crewai.flow/v1
        name: RoundTripFlow
        methods:
          begin:
            start: true
            do:
              call: code
              ref: test_flow_definition:RoundTripFlow.begin
          decide:
            listen: begin
            router: true
            do:
              call: code
              ref: test_flow_definition:RoundTripFlow.decide
          handle_left:
            listen: left
            do:
              call: code
              ref: test_flow_definition:RoundTripFlow.handle_left
        """,
    ]

    for declaration in declarations:
        loaded = flow_definition.FlowDefinition.from_declaration(contents=declaration)

        assert loaded.name == expected.name
        assert loaded.methods["decide"].router is True
        assert loaded.methods["decide"].listen == "begin"


def test_flow_definition_from_declaration_accepts_contents():
    data = {
        "schema": "crewai.flow/v1",
        "name": "DeclarationFlow",
        "methods": {
            "begin": {
                "start": True,
                "do": {
                    "call": "expression",
                    "expr": "'started'",
                },
            },
        },
    }
    definition = flow_definition.FlowDefinition.from_declaration(contents=data)
    contents = [
        definition,
        data,
        """
        {
          "schema": "crewai.flow/v1",
          "name": "DeclarationFlow",
          "methods": {
            "begin": {
              "start": true,
              "do": {
                "call": "expression",
                "expr": "'started'"
              }
            }
          }
        }
        """,
        """
        schema: crewai.flow/v1
        name: DeclarationFlow
        methods:
          begin:
            start: true
            do:
              call: expression
              expr: "'started'"
        """,
    ]

    for content in contents:
        loaded = flow_definition.FlowDefinition.from_declaration(contents=content)

        assert loaded.to_dict() == definition.to_dict()

def test_flow_definition_from_declaration_rejects_empty_file(tmp_path: Path):
    declaration_path = tmp_path / "flow.crewai"
    declaration_path.write_text(" \n", encoding="utf-8")

    with pytest.raises(ValueError, match="Flow declaration file is empty"):
        flow_definition.FlowDefinition.from_declaration(path=declaration_path)


@pytest.mark.parametrize("contents", ["[]", "false", "0", "null", "~"])
def test_flow_definition_from_declaration_rejects_falsey_non_mapping_contents(
    contents: str,
):
    with pytest.raises(ValueError, match="Flow declaration must contain a mapping"):
        flow_definition.FlowDefinition.from_declaration(contents=contents)


def test_flow_definition_from_declaration_accepts_paths(tmp_path: Path):
    definition = flow_definition.FlowDefinition.from_declaration(contents=
        {
            "schema": "crewai.flow/v1",
            "name": "DeclarationFlow",
            "methods": {
                "begin": {
                    "start": True,
                    "do": {
                        "call": "expression",
                        "expr": "'started'",
                    },
                },
            },
        }
    )
    declaration_path = tmp_path / "flow.crewai"
    declaration_path.write_text(
        """
        schema: crewai.flow/v1
        name: DeclarationFlow
        methods:
          begin:
            start: true
            do:
              call: expression
              expr: "'started'"
        """,
        encoding="utf-8",
    )
    path_inputs = [
        declaration_path,
        str(declaration_path),
    ]

    for path_input in path_inputs:
        loaded = flow_definition.FlowDefinition.from_declaration(path=path_input)

        assert loaded.name == definition.name
        assert loaded.methods["begin"].is_start is True
        assert loaded.methods["begin"].do.call == "expression"
        assert loaded.source_path == declaration_path.resolve()


def test_flow_definition_from_declaration_requires_input():
    with pytest.raises(ValueError, match="Provide contents or path"):
        flow_definition.FlowDefinition.from_declaration()


def test_flow_definition_from_declaration_prefers_contents_over_path(
    tmp_path: Path,
):
    data = {
        "schema": "crewai.flow/v1",
        "name": "ContentsFlow",
        "methods": {
            "begin": {
                "start": True,
                "do": {"call": "expression", "expr": "'started'"},
            },
        },
    }
    declaration_path = tmp_path / "missing.crewai"

    loaded = flow_definition.FlowDefinition.from_declaration(
        contents=data,
        path=declaration_path,
    )

    assert loaded.name == "ContentsFlow"
    assert loaded.source_path is None


def test_each_action_loads_from_declaration():
    definition = flow_definition.FlowDefinition.from_declaration(contents=
        {
            "schema": "crewai.flow/v1",
            "name": "EachFlow",
            "methods": {
                "process_rows": {
                    "description": "Process every loaded row.",
                    "start": True,
                    "do": {
                        "call": "each",
                        "in": "state.rows",
                        "do": [
                            {
                                "name": "normalize",
                                "action": {
                                    "call": "tool",
                                    "ref": "my_tools:NormalizeRowTool",
                                    "with": {"row": "${ item }"},
                                }
                            },
                            {
                                "name": "save",
                                "action": {
                                    "call": "code",
                                    "ref": "my_flow:save_row",
                                    "with": {
                                        "row": "${ item }",
                                        "normalized": "${ outputs.normalize }",
                                    },
                                }
                            },
                        ],
                    },
                }
            },
        }
    )

    assert definition.methods["process_rows"].description == "Process every loaded row."
    assert definition.methods["process_rows"].do.call == "each"


def test_flow_definition_rejects_invalid_method_names():
    with pytest.raises(ValueError, match="Flow method names must match"):
        flow_definition.FlowDefinition.from_declaration(contents=
            {
                "schema": "crewai.flow/v1",
                "name": "InvalidMethodNameFlow",
                "methods": {
                    "process-rows": {
                        "start": True,
                        "do": {
                            "call": "expression",
                            "expr": "'done'",
                        },
                    }
                },
            }
        )


def test_flow_definition_detects_persist_metadata():
    @persist(verbose=True)
    class PersistedFlow(Flow[dict]):
        initial_state = {}

        @start()
        def begin(self):
            return "started"

        @persist(verbose=False)
        @listen(begin)
        def checkpoint(self):
            return "saved"

    definition = PersistedFlow.flow_definition()

    assert definition.persist is not None
    assert definition.persist.enabled is True
    assert definition.persist.verbose is True

    assert definition.methods["begin"].persist is None

    method_persist = definition.methods["checkpoint"].persist
    assert method_persist is not None
    assert method_persist.enabled is True
    assert method_persist.verbose is False


def test_flow_definition_allows_dynamic_router_emit():
    class DynamicRouterFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def decide(self):
            return self.state["dynamic_event"]

    definition = DynamicRouterFlow.flow_definition()

    assert definition.methods["decide"].emit is None


def test_flow_definition_infers_literal_router_emit():
    class LiteralRouterFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def decide(self) -> Literal["left", "right"]:
            return "left"

        @listen("left")
        def handle_left(self):
            return "left"

        @listen("right")
        def handle_right(self):
            return "right"

    definition = LiteralRouterFlow.flow_definition()

    assert definition.methods["decide"].emit == ["left", "right"]


def test_flow_definition_infers_enum_router_emit():
    class Decision(str, Enum):
        APPROVE = "approve"
        REJECT = "reject"

    class EnumRouterFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def decide(self) -> Decision:
            return Decision.APPROVE

        @listen("approve")
        def handle_approve(self):
            return "approve"

        @listen("reject")
        def handle_reject(self):
            return "reject"

    definition = EnumRouterFlow.flow_definition()

    assert definition.methods["decide"].emit == ["approve", "reject"]


def test_flow_definition_infers_literal_union_router_emit():
    class LiteralUnionRouterFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def decide(self) -> Literal["left"] | Literal["right"]:
            return "left"

        @listen("left")
        def handle_left(self):
            return "left"

        @listen("right")
        def handle_right(self):
            return "right"

    definition = LiteralUnionRouterFlow.flow_definition()

    assert definition.methods["decide"].emit == ["left", "right"]


def test_flow_definition_infers_annotated_literal_router_emit():
    class AnnotatedRouterFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def decide(self) -> Annotated[Literal["left"] | None, "route"]:
            return "left"

    definition = AnnotatedRouterFlow.flow_definition()

    assert definition.methods["decide"].emit == ["left"]


def test_flow_definition_does_not_infer_container_literal_router_emit():
    class ContainerLiteralRouterFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def list_route(self) -> list[Literal["left"]]:
            return ["left"]

        @router(begin)
        def dict_route(self) -> dict[str, Literal["right"]]:
            return {"route": "right"}

    definition = ContainerLiteralRouterFlow.flow_definition()

    assert definition.methods["list_route"].emit is None
    assert definition.methods["dict_route"].emit is None


def test_flow_definition_does_not_infer_unannotated_router_body_emit():
    class UnannotatedRouterFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def decide(self):
            return "left"

        @listen("left")
        def handle_left(self):
            return "left"

    definition = UnannotatedRouterFlow.flow_definition()

    assert definition.methods["decide"].emit is None


def test_flow_definition_accepts_explicit_router_events():
    class ExplicitRouterFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin, emit=["left", "right", "left"])
        def decide(self):
            return self.state["dynamic_event"]

        @listen("left")
        def handle_left(self):
            return "left"

        @listen("right")
        def handle_right(self):
            return "right"

    definition = ExplicitRouterFlow.flow_definition()

    assert definition.methods["decide"].emit == ["left", "right"]


def test_flow_definition_ignores_legacy_diagnostics_loaded_from_contract():
    definition = flow_definition.FlowDefinition.from_declaration(contents=
        {
            "schema": "crewai.flow/v1",
            "name": "LoadedDiagnosticsFlow",
            "methods": {
                "begin": {
                    "do": {"ref": "loaded_flows:LoadedDiagnosticsFlow.begin"},
                    "start": True,
                }
            },
            "diagnostics": [
                {
                    "code": "serialized_warning",
                    "message": "Preserved serialized diagnostic",
                    "severity": "warning",
                    "path": "methods.decision",
                },
                {
                    "code": "router_without_trigger",
                    "message": "router: true requires either start or listen",
                    "severity": "error",
                    "path": "methods.decision",
                },
            ],
        }
    )

    assert "diagnostics" not in definition.to_dict()


def test_router_start_false_without_listen_is_allowed(caplog):
    caplog.set_level(logging.ERROR, logger="crewai.flow.flow_definition")

    flow_definition.FlowDefinition.from_declaration(contents=
        {
            "schema": "crewai.flow/v1",
            "name": "LoadedFlow",
            "methods": {
                "decision": {
                    "do": {"ref": "loaded_flows:LoadedFlow.decision"},
                    "router": True,
                    "start": False,
                    "emit": ["continue"],
                }
            },
        }
    )

    assert not caplog.records


def test_router_human_feedback_preserves_existing_router_metadata():
    class RouterHumanFeedbackFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @human_feedback(message="Review route:")
        @router(begin, emit=["approved", "rejected"])
        def decide(self):
            return "approved"

        @listen("approved")
        def handle_approved(self):
            return "approved"

    definition = RouterHumanFeedbackFlow.flow_definition()
    method = definition.methods["decide"]

    assert method.router is True
    assert method.listen == "begin"
    assert method.emit == ["approved", "rejected"]
    assert method.human_feedback is not None


def test_dynamic_router_flow_definition_allows_dynamic_emit():
    class LazyDynamicRouterFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def decide(self):
            return self.state["dynamic_event"]

    definition = LazyDynamicRouterFlow.flow_definition()
    assert definition.methods["decide"].emit is None


def test_dynamic_router_string_listener_is_valid_contract():
    class DynamicRouterListenerFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @router(begin)
        def decide(self):
            return self.state["dynamic_event"]

        @listen("dynamic_event")
        def handle(self):
            return "handled"

    definition = DynamicRouterListenerFlow.flow_definition()

    assert definition.methods["handle"].listen == "dynamic_event"


def test_static_string_listener_is_allowed_by_contract():
    definition = flow_definition.FlowDefinition.from_declaration(contents=
        {
            "schema": "crewai.flow/v1",
            "name": "TypoFlow",
            "methods": {
                "begin": {
                    "do": {"ref": "loaded_flows:TypoFlow.begin"},
                    "start": True,
                },
                "handle": {
                    "do": {"ref": "loaded_flows:TypoFlow.handle"},
                    "listen": "begni",
                },
            },
        }
    )
    assert definition.methods["handle"].listen == "begni"


@pytest.mark.parametrize("listen", ["publish", {"or": ["publish", "revise"]}])
@pytest.mark.parametrize("router_enabled", [False, True])
def test_flow_definition_rejects_method_self_listen(listen, router_enabled):
    with pytest.raises(ValueError, match="listen condition"):
        flow_definition.FlowDefinition.from_declaration(contents=
            {
                "schema": "crewai.flow/v1",
                "name": "SelfListenFlow",
                "methods": {
                    "begin": {
                        "do": {"ref": "loaded_flows:SelfListenFlow.begin"},
                        "start": True,
                    },
                    "publish": {
                        "do": {"ref": "loaded_flows:SelfListenFlow.publish"},
                        "listen": listen,
                        "router": router_enabled,
                        "emit": ["done"] if router_enabled else None,
                    },
                },
            }
        )


def test_flow_definition_rejects_conversational_route_handler_name_collision():
    with pytest.raises(ValueError, match=r"listen condition 'create_video'"):
        flow_definition.FlowDefinition.from_declaration(contents=
            {
                "schema": "crewai.flow/v1",
                "name": "VideoFlow",
                "conversational": {
                    "enabled": True,
                    "router": {
                        "route_descriptions": {
                            "create_video": "User wants a new video.",
                        },
                    },
                },
                "methods": {
                    "begin": {
                        "do": {"ref": "loaded_flows:VideoFlow.begin"},
                        "start": True,
                    },
                    "create_video": {
                        "do": {"ref": "loaded_flows:VideoFlow.create_video"},
                        "listen": "create_video",
                    },
                },
            }
        )


def test_build_flow_definition_wraps_validation_error_with_class_name():
    class VideoFlow(Flow):
        conversational = True

        @listen("create_video")
        def create_video(self):
            return "made a video"

    with pytest.raises(ValueError, match="Invalid flow definition for VideoFlow"):
        VideoFlow.flow_definition()

    with pytest.raises(ValueError, match="Invalid flow definition for VideoFlow"):
        VideoFlow()


def test_start_false_not_classified_as_start_method():
    definition = flow_definition.FlowDefinition.from_declaration(contents=
        {
            "schema": "crewai.flow/v1",
            "name": "ExplicitNonStartFlow",
            "methods": {
                "begin": {
                    "do": {"ref": "loaded_flows:ExplicitNonStartFlow.begin"},
                    "start": True,
                },
                "handle": {
                    "do": {"ref": "loaded_flows:ExplicitNonStartFlow.handle"},
                    "start": False,
                    "listen": "begin",
                },
            },
        }
    )

    assert definition.methods["begin"].is_start is True
    assert definition.methods["handle"].is_start is False

    class ExplicitNonStartFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @listen(begin)
        def handle(self):
            return "handled"

    # Attach the loaded contract (with explicit ``start: false``) so the
    # projections read from it rather than rebuilding from the DSL.
    ExplicitNonStartFlow._flow_definition = definition

    flow = ExplicitNonStartFlow()
    viz_structure = visualization_builder.build_flow_structure(flow)
    assert "handle" not in viz_structure["start_methods"]
    assert viz_structure["nodes"]["handle"]["type"] != "start"


def test_flow_definition_cache_is_not_reused_by_subclasses():
    class ParentFlow(Flow):
        @start()
        def begin(self):
            return "begin"

    parent_definition = ParentFlow.flow_definition()

    class ChildFlow(ParentFlow):
        @listen(ParentFlow.begin)
        def child_step(self):
            return "child"

    child_definition = ChildFlow.flow_definition()

    assert parent_definition.name == "ParentFlow"
    assert child_definition.name == "ChildFlow"
    assert child_definition is not parent_definition
    assert set(child_definition.methods) == {"child_step"}


def test_flow_definition_allows_router_without_trigger(caplog):
    caplog.set_level(logging.WARNING, logger="crewai.flow.flow_definition")

    flow_definition.FlowDefinition.from_declaration(contents=
        {
            "schema": "crewai.flow/v1",
            "name": "LoadedFlow",
            "methods": {
                "decision": {
                    "do": {"ref": "loaded_flows:LoadedFlow.decision"},
                    "router": True,
                    "emit": ["continue"],
                }
            },
        }
    )

    class StandaloneRouterFlow(Flow):
        @router(emit=["continue"])
        def decision(self):
            return "continue"

    StandaloneRouterFlow.flow_definition()

    assert not caplog.records


def test_skill_documents_flow_wiring():
    skill = flow_definition.FlowDefinition.skill()

    assert isinstance(skill, str)
    assert "```yaml" in skill
    assert "[Method](#method-methods)" in skill
    assert 'input: "Reviewed research: ${outputs.research_brief.raw}"' in skill
    assert "do not assemble the string with CEL `+`" in skill
    assert "Do not use CEL `+` to build text in action mappings" in skill
    assert "Agent prompt template. Insert Flow values with `${...}`" in skill
    assert (
        "Repository-backed agents may set `from_repository` and omit inline "
        "`role`, `goal`, and `backstory`" in skill
    )
    assert "Runtime inputs passed to the Crew" in skill
    assert "Tool input arguments. Insert Flow values with `${...}`" in skill
    assert "trust CrewAI defaults and omit them" in skill
    assert "#### LLM Definition" in skill
    assert "`max_tokens` (optional): integer | null; default `null`" in skill
    assert "CrewAI does not set an explicit output token cap" in skill
    assert "`planning_config` (optional): object | null; default `null`" in skill
    assert "Set `max_attempts` to limit planning refinement attempts" in skill
    assert "`allow_delegation` (optional): boolean | null; default `null`" in skill
    assert "`max_iter` (optional): integer | null; default `null`" in skill
    assert "`max_rpm` (optional): integer | null; default `null`" in skill
    assert "`max_execution_time` (optional): integer | null; default `null`" in skill
    assert "Maximum execution time in seconds for an agent" in skill
    for rule in FLOW_TEMPLATE_EXPRESSION_RULES:
        assert rule in skill
    for example in FLOW_TEMPLATE_EXPRESSION_EXAMPLES["yaml"]:
        assert example["title"] in skill
        assert example["code"] in skill


def test_skill_renders_both_conversational_sections():
    """Both models must render; the router shares no section with its parent.

    Non-union sections render only their first model, so grouping them would
    drop the router and leave the link to it without a target.
    """
    skill = flow_definition.FlowDefinition.skill()

    assert "### Conversational (`conversational`)" in skill
    assert "### Conversational Router (`conversational.router`)" in skill

    link = "[Conversational Router (`conversational.router`)]"
    assert link in skill
    anchor = re.search(re.escape(link) + r"\((#[^)]+)\)", skill).group(1)
    assert anchor == "#conversational-router-conversationalrouter"


def test_skill_documents_every_conversational_field():
    skill = flow_definition.FlowDefinition.skill()
    section = skill[skill.index("### Conversational (`conversational`)") :]

    for field in flow_definition.FlowConversationalDefinition.model_fields:
        assert f"`{field}`" in section, field
    for field in flow_definition.FlowConversationalRouterDefinition.model_fields:
        assert f"`{field}`" in section, field


def test_skill_conversational_skip_suppresses_both_sections():
    skill = flow_definition.FlowDefinition.skill(skips=["conversational"])

    assert "### Conversational" not in skill
    assert "conversational-router" not in skill


def test_skill_can_render_json_examples():
    skill = flow_definition.FlowDefinition.skill(examples_format="json")

    assert "```json" in skill
    assert '"schema": "crewai.flow/v1"' in skill
    for example in FLOW_TEMPLATE_EXPRESSION_EXAMPLES["json"]:
        assert example["title"] in skill
        assert example["code"] in skill
    assert FLOW_TEMPLATE_EXPRESSION_EXAMPLES["yaml"][0]["code"] not in skill
    assert "```yaml" not in skill


def test_skill_ignores_unknown_skips():
    skill = flow_definition.FlowDefinition.skill(skips=["unknown"])

    assert "[Method](#method-methods)" in skill


def test_skill_with_skips_is_shorter():
    full = flow_definition.FlowDefinition.skill()
    trimmed = flow_definition.FlowDefinition.skill(
        skips=[
            "each",
            "hitl",
            "persistence",
            "expression_action",
            "script_action",
            "tool_action",
        ]
    )

    assert "[Method](#method-methods)" in trimmed
    assert "call: expression" not in trimmed
    assert "Prefer `call: expression`" not in trimmed
    assert "call: script" not in trimmed
    assert "call: tool" not in trimmed
    assert len(trimmed) < len(full)