Merge branch 'main' into lorenze/imp/canary-flow-deployment-e2e-tests

docs/edge/en/guides/flows/conversational-flows.mdx

@@ -376,6 +376,27 @@ def handle_internet_search(self) -> str:
     ...
 ```
 
+### Naming handlers
+
+The string in `@listen("…")` is a **router route label** (an event name), not the Python method name. Route labels and method completion events share one trigger namespace, so naming a handler the same as its route causes the handler to re-trigger itself in a loop.
+
+Use a different method name — the docs examples use a `handle_*` prefix:
+
+```python
+@listen("create_video")
+def handle_create_video(self) -> str:
+    """User wants a new video."""
+    ...
+```
+
+Do **not** mirror the route label on the method:
+
+```python
+@listen("create_video")
+def create_video(self) -> str:  # rejected at flow instantiation
+    ...
+```
+
 …and the router LLM sees:
 
 ```

lib/cli/src/crewai_cli/templates/AGENTS.md

@@ -627,6 +627,26 @@ class MyFlow(Flow):
 | `@listen(method)` | Triggers when specified method completes. Receives output as argument |
 | `@router(method)` | Conditional branching. Returns string labels that trigger `@listen("label")` |
 
+### `@listen` labels vs handler names
+
+The string in `@listen("...")` is an **event or route label**, not the Python method name. Router return values, route labels, and method completion events share one trigger namespace.
+
+**Never** use the same name for the `@listen` label and the handler method:
+
+```python
+# ❌ Wrong — raises a validation error when the flow is instantiated
+@listen("create_video")
+def create_video(self):
+    ...
+
+# ✅ Correct — distinct handler name (handle_* prefix is a common pattern)
+@listen("create_video")
+def handle_create_video(self):
+    ...
+```
+
+If validation were bypassed, matching names would also cause the handler to re-trigger itself in a loop at runtime. This applies to all flows. It is especially common in **conversational flows** (`conversational = True`), where `@listen("...")` is a router intent name — do not name the handler after the route it serves.
+
 ### Structured State
 ```python
 from pydantic import BaseModel
@@ -1148,3 +1168,4 @@ crewai run                    # Execute
 - Using `process=Process.hierarchical` without setting `manager_llm` or `manager_agent`
 - Circular delegation: set `allow_delegation=False` on specialist agents
 - Not installing tools package: `uv add crewai-tools`
+- **Matching `@listen("label")` to the handler method name** — raises a validation error at flow instantiation; would re-trigger in an infinite loop at runtime only if validation is bypassed. Use a different method name (e.g. `handle_create_video` for `@listen("create_video")`)

lib/cli/src/crewai_cli/templates/declarative_flow/AGENTS.md

@@ -39,8 +39,8 @@ Pick the simplest action that does the job.
 - `state` is the initial shared data shape. Action results do not automatically merge into `state`.
 - Read method results with `outputs.method_name` after that method can run.
 - `listen` targets a method name or a router-emitted event name.
-- Methods must not listen to their own method name.
-- Method names and emitted event names share one namespace. Avoid reusing the same string for both unless the user explicitly wants that.
+- Methods must not listen to their own method name — including when the `listen` value is a route label that matches the method name (e.g. `listen: create_video` on method `create_video`).
+- Method names and emitted event names share one namespace. Do not reuse the same string for a method's `listen` target and its method name.
 - Use `router: true` plus `emit` when one method chooses between named branches.
 - A router action must return exactly one emitted event string. It must not return JSON, a list, or an explanation.
 - Use `start: true` for the single entrypoint.
@@ -107,8 +107,8 @@ Dynamic value rules:
 - Do not make `do` a list.
 - Do not use CEL `+` to build text in action mappings. Keep the text literal and insert each dynamic value with `${...}`.
 - Do not reference `outputs.some_method` before `some_method` can run.
-- Do not set a method's `listen` to its own method name.
-- Do not use the same string for an emitted event and a method name unless the user asks for it.
+- Do not set a method's `listen` to its own method name (including matching route labels such as `listen: create_video` on method `create_video`).
+- Do not use the same string for a method's `listen` target and its method name.
 - Do not use `emit` without `router: true`.
 - Do not rely on crew action-level `inputs` alone to ground agent behavior. Inputs that do not match placeholders are effectively unused by the prompt.
 - Do not ask agents to infer missing facts when accuracy matters. Tell them to mark missing dates, amounts, offers, logs, or constraints as unknown.

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

@@ -4,7 +4,7 @@ import json
 import logging
 from typing import Any, ParamSpec, TypeVar
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 from typing_extensions import TypeIs
 
 from crewai.flow.flow_definition import (
@@ -432,6 +432,20 @@ def _iter_flow_methods(flow_class: type) -> dict[str, Any]:
     return methods
 
 
+def _flow_definition_validation_error(
+    flow_class: type, exc: ValidationError
+) -> ValueError:
+    errors = exc.errors()
+    if errors:
+        detail = errors[0].get("msg", str(exc))
+        if isinstance(detail, str) and detail.startswith("Value error, "):
+            detail = detail.removeprefix("Value error, ")
+    else:
+        detail = str(exc)
+    class_name = getattr(flow_class, "__name__", "Flow")
+    return ValueError(f"Invalid flow definition for {class_name}: {detail}")
+
+
 def _build_flow_definition_from_class(
     flow_class: type,
     namespace: dict[str, Any] | None = None,
@@ -455,15 +469,18 @@ def _build_flow_definition_from_class(
     if docstring:
         description = docstring.strip()
 
-    definition = FlowDefinition(
-        name=getattr(flow_class, "__name__", "Flow"),
-        description=description,
-        state=_build_state_definition(flow_class),
-        config=_build_config_definition(flow_class),
-        persist=_build_persistence_definition(flow_class),
-        conversational=_build_conversational_definition(flow_class),
-        methods=methods,
-    )
+    try:
+        definition = FlowDefinition(
+            name=getattr(flow_class, "__name__", "Flow"),
+            description=description,
+            state=_build_state_definition(flow_class),
+            config=_build_config_definition(flow_class),
+            persist=_build_persistence_definition(flow_class),
+            conversational=_build_conversational_definition(flow_class),
+            methods=methods,
+        )
+    except ValidationError as exc:
+        raise _flow_definition_validation_error(flow_class, exc) from exc
     log_flow_definition_issues(definition)
     return definition
 

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

@@ -775,7 +775,11 @@ class FlowDefinition(BaseModel):
         for method_name, method in self.methods.items():
             if _condition_references(method.listen, method_name):
                 raise ValueError(
-                    f"methods.{method_name}.listen must not reference itself"
+                    _self_listen_error(
+                        method_name=method_name,
+                        listen=method.listen,
+                        definition=self,
+                    )
                 )
         return self
 
@@ -888,6 +892,39 @@ def _condition_references(condition: FlowDefinitionCondition | None, name: str)
     )
 
 
+def _format_listen_condition(condition: FlowDefinitionCondition | None) -> str:
+    if condition is None:
+        return "None"
+    return repr(condition)
+
+
+def _self_listen_error(
+    *,
+    method_name: str,
+    listen: FlowDefinitionCondition | None,
+    definition: FlowDefinition,
+) -> str:
+    path = f"methods.{method_name}.listen"
+    listen_display = _format_listen_condition(listen)
+    conversational = (
+        definition.conversational is not None and definition.conversational.enabled
+    )
+    if conversational:
+        return (
+            f"{path} listen condition {listen_display} matches the handler name "
+            f"{method_name!r}. In conversational flows, @listen labels are router "
+            "route names — they share the same trigger namespace as method completion "
+            "events, so this handler would re-run in a loop. Rename the handler "
+            f"(for example, handle_{method_name}) or use a different route label."
+        )
+
+    return (
+        f"{path} listen condition {listen_display} references the handler name "
+        f"{method_name!r}. A listener triggered by its own completion creates an "
+        "infinite loop. Listen to a different method or event, or rename the handler."
+    )
+
+
 def _validate_action_cel(
     action: FlowActionDefinition,
     *,

lib/crewai/tests/test_flow.py

@@ -2158,7 +2158,7 @@ def test_self_listening_method_is_rejected():
         def process(self):
             pass
 
-    with pytest.raises(ValueError, match="methods.process.listen"):
+    with pytest.raises(ValueError, match="Invalid flow definition for SelfListenFlow"):
         SelfListenFlow.flow_definition()
 
 
@@ -2176,7 +2176,7 @@ def test_or_condition_self_listen_is_rejected():
         def process(self):
             pass
 
-    with pytest.raises(ValueError, match="methods.process.listen"):
+    with pytest.raises(ValueError, match="Invalid flow definition for OrSelfListenFlow"):
         OrSelfListenFlow.flow_definition()
 
 
@@ -2190,7 +2190,7 @@ def test_router_self_listening_method_is_rejected():
         def route(self):
             return "done"
 
-    with pytest.raises(ValueError, match="methods.route.listen"):
+    with pytest.raises(ValueError, match="Invalid flow definition for RouterSelfListenFlow"):
         RouterSelfListenFlow.flow_definition()
 
 

lib/crewai/tests/test_flow_definition.py

@@ -1231,7 +1231,7 @@ def test_static_string_listener_is_allowed_by_contract():
 @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="methods.publish.listen"):
+    with pytest.raises(ValueError, match="listen condition"):
         flow_definition.FlowDefinition.from_declaration(contents=
             {
                 "schema": "crewai.flow/v1",
@@ -1252,6 +1252,49 @@ def test_flow_definition_rejects_method_self_listen(listen, router_enabled):
         )
 
 
+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=
         {