[EPD-197] Warn locally when a persisted flow gets a non-UUID id input (AMP parity)

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 = (

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"