fix: ensure persisted state overrides class defaults

- Remove early return in Flow.__init__ to allow proper state initialization
- Add test_flow_default_override.py to verify state override behavior
- Fix issue where default values weren't being overridden by persisted state

Fixes the issue where persisted state values weren't properly overriding
class defaults when restarting a flow with a previously saved state ID.

Co-Authored-By: Joe Moura <joao@crewai.com>

src/crewai/flow/flow.py

@@ -499,12 +499,8 @@ class Flow(Generic[T], metaclass=FlowMeta):
             elif kwargs and "id" in kwargs:
                 self._log_flow_event(f"Loading flow state from memory for ID: {kwargs['id']}", color="bold_yellow")
                 stored_state = self._persistence.load_state(kwargs["id"])
-                if not stored_state:
-                    # For kwargs["id"], we allow creating new state if not found
-                    self._state = self._create_initial_state()
-                    if kwargs:
-                        self._initialize_state(kwargs)
-                    return
+                # Don't return early if state not found - let the normal initialization flow handle it
+                # This ensures proper state initialization and override behavior
 
         # Initialize state based on persistence and kwargs
         if stored_state:

tests/test_flow_default_override.py

@@ -0,0 +1,53 @@
+"""Test that persisted state properly overrides default values."""
+
+import os
+from typing import Optional
+
+import pytest
+from pydantic import BaseModel
+
+from crewai.flow.flow import Flow, FlowState, start
+from crewai.flow.persistence import persist
+from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
+
+
+class PoemState(FlowState):
+    """Test state model with default values that should be overridden."""
+    sentence_count: int = 1000  # Default that should be overridden
+    poem: str = ""
+
+
+def test_default_value_override(tmp_path):
+    """Test that persisted state values override class defaults."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+    
+    @persist(persistence)
+    class PoemFlow(Flow[PoemState]):
+        initial_state = PoemState
+        
+        @start()
+        def set_sentence_count(self):
+            # First run will set a custom value
+            if not self.state.sentence_count or self.state.sentence_count == 1000:
+                self.state.sentence_count = 2
+    
+    # First run - should set sentence_count to 2
+    # First run - should set sentence_count to 2
+    flow1 = PoemFlow(persistence=persistence)
+    flow1.kickoff()
+    original_uuid = flow1.state.id
+    assert flow1.state.sentence_count == 2
+    
+    # Second run - should load sentence_count=2 instead of default 1000
+    flow2 = PoemFlow(persistence=persistence)
+    flow2.kickoff(inputs={"id": original_uuid})
+    assert flow2.state.sentence_count == 2  # Should load 2, not default 1000
+    
+    # Third run - explicit override should work
+    flow3 = PoemFlow(
+        persistence=persistence,
+        id=original_uuid,
+        sentence_count=3,  # Override persisted value
+    )
+    assert flow3.state.sentence_count == 3  # Should use override value