Also warn for method-level @persist flows; drop unused pytest import

Method-level @persist stamps persistence into the flow definition's
per-method config and never sets self.persistence, so the AMP parity
warning missed flows persisting only via method decorators. Detect
active persistence across instance/class-level backends, the flow-level
definition, and per-method persist configs.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -2086,7 +2086,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         try:
             # AMP parity: persisted flows deployed to CrewAI AMP reject
             # non-UUID `id` inputs with HTTP 422, so surface that locally.
-            if inputs and "id" in inputs and self.persistence is not None:
+            # 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
@@ -2692,6 +2694,23 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     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

@@ -10,7 +10,6 @@ import os
 import warnings
 from uuid import uuid4
 
-import pytest
 from crewai.flow.flow import Flow, start
 from crewai.flow.persistence import persist
 from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
@@ -77,6 +76,35 @@ def test_valid_uuid_id_with_persistence_does_not_warn(tmp_path):
     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."""