diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index a6c73e6979..4c8935644a 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -30,7 +30,8 @@ TypeVar, cast, ) -from uuid import uuid4 +from uuid import UUID, uuid4 +import warnings from opentelemetry import baggage from opentelemetry.context import attach, detach @@ -161,6 +162,27 @@ logger = logging.getLogger(__name__) +def _warn_if_non_uuid_flow_state_id(value: Any) -> None: + """Warn once per kickoff when a persisted flow gets a non-UUID `id` input. + + CrewAI AMP deployments reject non-UUID `inputs.id` values with HTTP 422, + while the OSS library accepts any string locally. Surfacing the mismatch + here keeps local development aligned with deployed behavior without + changing how the flow runs. + """ + try: + UUID(str(value)) + except (ValueError, TypeError): + warnings.warn( + f"Flow state id {value!r} is not a valid UUID. Non-UUID flow state " + "ids are rejected by CrewAI AMP deployments with HTTP 422; use a " + "UUID (e.g. str(uuid4())) for the 'id' input to keep local and " + "deployed behavior aligned.", + UserWarning, + stacklevel=2, + ) + + def _condition_branches( condition: dict[str, Any], ) -> tuple[Literal["and", "or"], list[FlowDefinitionCondition]]: @@ -2062,6 +2084,13 @@ async def kickoff_async( self._attach_usage_aggregation_listener() try: + # AMP parity: persisted flows deployed to CrewAI AMP reject + # non-UUID `id` inputs with HTTP 422, so surface that locally. + # Covers instance/class-level persistence and method-level + # @persist (which never sets self.persistence). + if inputs and "id" in inputs and self._has_active_persistence(): + _warn_if_non_uuid_flow_state_id(inputs["id"]) + # Reset flow state for fresh execution unless restoring from persistence is_restoring = ( inputs and "id" in inputs and self.persistence is not None @@ -2665,6 +2694,23 @@ async def _execute_method( self._event_futures.append(future) raise e + def _has_active_persistence(self) -> bool: + """Whether this flow persists state anywhere. + + True when an instance/class-level backend is set, the flow-level + definition enables persistence, or any method carries a method-level + ``@persist`` (which never sets ``self.persistence``). + """ + if self.persistence is not None: + return True + flow_persist = self._definition.persist + if flow_persist is not None and flow_persist.enabled: + return True + return any( + method.persist is not None and method.persist.enabled + for method in self._definition.methods.values() + ) + def _persist_method_completion(self, method_name: FlowMethodName) -> None: method_definition = self._definition.methods[str(method_name)] persist_definition = ( diff --git a/lib/crewai/tests/test_flow_id_format_warning.py b/lib/crewai/tests/test_flow_id_format_warning.py new file mode 100644 index 0000000000..6952c4d46b --- /dev/null +++ b/lib/crewai/tests/test_flow_id_format_warning.py @@ -0,0 +1,129 @@ +"""Tests for the non-UUID flow state id warning (CrewAI AMP parity). + +CrewAI AMP deployments reject non-UUID ``inputs.id`` values with HTTP 422, +while the OSS library accepts any string locally. A persisted flow kicked off +with a non-UUID ``id`` should surface a one-time warning so the mismatch is +caught before deployment — without changing local behavior. +""" + +import os +import warnings +from uuid import uuid4 + +from crewai.flow.flow import Flow, start +from crewai.flow.persistence import persist +from crewai.flow.persistence.sqlite import SQLiteFlowPersistence + + +AMP_WARNING_MATCH = "CrewAI AMP" + + +def _build_persisted_flow_class(persistence): + class PersistedFlow(Flow[dict]): + initial_state = dict + + @start() + @persist(persistence) + def init_step(self): + self.state["message"] = "ran" + + return PersistedFlow + + +def test_non_uuid_id_with_persistence_warns_once_and_still_runs(tmp_path): + """A non-UUID `id` input on a persisted flow warns once but runs fine.""" + db_path = os.path.join(tmp_path, "test_flows.db") + persistence = SQLiteFlowPersistence(db_path) + flow = _build_persisted_flow_class(persistence)(persistence=persistence) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + flow.kickoff(inputs={"id": "session-1"}) + + amp_warnings = [ + w for w in caught if AMP_WARNING_MATCH in str(w.message) + ] + assert len(amp_warnings) == 1, ( + f"expected exactly one AMP parity warning, got {len(amp_warnings)}" + ) + assert issubclass(amp_warnings[0].category, UserWarning) + message = str(amp_warnings[0].message) + assert "422" in message + assert "UUID" in message + + # Behavior is unchanged: the flow ran and kept the non-UUID id locally. + assert flow.state["id"] == "session-1" + assert flow.state["message"] == "ran" + assert persistence.load_state("session-1") is not None + + +def test_valid_uuid_id_with_persistence_does_not_warn(tmp_path): + """A valid UUID `id` input on a persisted flow emits no AMP warning.""" + db_path = os.path.join(tmp_path, "test_flows.db") + persistence = SQLiteFlowPersistence(db_path) + flow = _build_persisted_flow_class(persistence)(persistence=persistence) + valid_id = str(uuid4()) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + flow.kickoff(inputs={"id": valid_id}) + + amp_warnings = [ + w for w in caught if AMP_WARNING_MATCH in str(w.message) + ] + assert amp_warnings == [] + assert flow.state["id"] == valid_id + assert flow.state["message"] == "ran" + + +def test_non_uuid_id_with_method_level_persist_only_warns(tmp_path): + """Method-level @persist (no instance/class persistence) still warns. + + Method-level @persist never sets `flow.persistence`, but the flow still + saves state under the provided id, so the AMP parity warning must fire. + """ + db_path = os.path.join(tmp_path, "test_flows.db") + persistence = SQLiteFlowPersistence(db_path) + # No persistence= constructor arg: only the method is persisted. + flow = _build_persisted_flow_class(persistence)() + assert flow.persistence is None + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + flow.kickoff(inputs={"id": "session-1"}) + + amp_warnings = [ + w for w in caught if AMP_WARNING_MATCH in str(w.message) + ] + assert len(amp_warnings) == 1, ( + f"expected exactly one AMP parity warning, got {len(amp_warnings)}" + ) + + # Behavior is unchanged: the method-level @persist saved the state. + assert flow.state["id"] == "session-1" + assert flow.state["message"] == "ran" + assert persistence.load_state("session-1") is not None + + +def test_non_uuid_id_without_persistence_does_not_warn(): + """Without persistence, a non-UUID `id` input emits no AMP warning.""" + + class PlainFlow(Flow[dict]): + initial_state = dict + + @start() + def init_step(self): + self.state["message"] = "ran" + + flow = PlainFlow() + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + flow.kickoff(inputs={"id": "session-1"}) + + amp_warnings = [ + w for w in caught if AMP_WARNING_MATCH in str(w.message) + ] + assert amp_warnings == [] + assert flow.state["id"] == "session-1" + assert flow.state["message"] == "ran"