diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index a76a5bace..55ac5524e 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -1878,6 +1878,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): "is configured for this flow." ) + @staticmethod + def _check_restore_conflict( + from_checkpoint: CheckpointConfig | None, + restore_from_state_id: str | None, + ) -> None: + """Reject combining ``from_checkpoint`` with ``restore_from_state_id``.""" + if from_checkpoint is not None and restore_from_state_id is not None: + raise ValueError( + "Cannot combine `from_checkpoint` and `restore_from_state_id`. " + "These parameters target different state systems " + "(Checkpointing and @persist) and cannot be used together." + ) + def _ensure_restorable_state( self, restore_from_state_id: str | None, @@ -1888,14 +1901,17 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): Streaming kickoffs return a session before the runtime evaluates ``restore_from_state_id``; this pre-check keeps ``raise_on_missing_state=True`` failing at call time, matching the - non-streaming path. The runtime still performs the authoritative load - (and raises again) during execution. + non-streaming path. It records the miss in ``last_restore_succeeded`` + before raising, exactly like the runtime path. The runtime still + performs the authoritative load (and raises again) during execution. """ if not raise_on_missing_state or restore_from_state_id is None: return if self.persistence is None: + self._last_restore_succeeded = False raise self._no_persistence_error(restore_from_state_id) if not self.persistence.load_state(restore_from_state_id): + self._last_restore_succeeded = False raise self._missing_state_error(restore_from_state_id) def stream_events( @@ -1912,6 +1928,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): lookup miss raises ``ValueError`` here, before the session is returned, so callers do not have to consume frames to see the error. """ + self._check_restore_conflict(from_checkpoint, restore_from_state_id) self._ensure_restorable_state(restore_from_state_id, raise_on_missing_state) result_holder: list[Any] = [] state = create_frame_streaming_state(result_holder, use_async=False) @@ -1953,6 +1970,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): lookup miss raises ``ValueError`` here, before the session is returned, so callers do not have to consume frames to see the error. """ + self._check_restore_conflict(from_checkpoint, restore_from_state_id) self._ensure_restorable_state(restore_from_state_id, raise_on_missing_state) result_holder: list[Any] = [] state = create_frame_streaming_state(result_holder, use_async=True) @@ -2016,12 +2034,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): Returns: The final output from the flow or StreamSession if streaming. """ - if from_checkpoint is not None and restore_from_state_id is not None: - raise ValueError( - "Cannot combine `from_checkpoint` and `restore_from_state_id`. " - "These parameters target different state systems " - "(Checkpointing and @persist) and cannot be used together." - ) + self._check_restore_conflict(from_checkpoint, restore_from_state_id) restored = apply_checkpoint(self, from_checkpoint) if restored is not None: return restored.kickoff(inputs=inputs, input_files=input_files) @@ -2091,12 +2104,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): Returns: The final output from the flow, which is the result of the last executed method. """ - if from_checkpoint is not None and restore_from_state_id is not None: - raise ValueError( - "Cannot combine `from_checkpoint` and `restore_from_state_id`. " - "These parameters target different state systems " - "(Checkpointing and @persist) and cannot be used together." - ) + self._check_restore_conflict(from_checkpoint, restore_from_state_id) restored = apply_checkpoint(self, from_checkpoint) if restored is not None: return await restored.kickoff_async(inputs=inputs, input_files=input_files) diff --git a/lib/crewai/tests/test_flow_persistence.py b/lib/crewai/tests/test_flow_persistence.py index 9fcce5c3d..83eebd7c3 100644 --- a/lib/crewai/tests/test_flow_persistence.py +++ b/lib/crewai/tests/test_flow_persistence.py @@ -513,6 +513,96 @@ async def test_astream_strict_restore_miss_raises_eagerly(tmp_path): assert flow2.state.counter == 0 +def test_stream_eager_strict_miss_sets_last_restore_succeeded_false(tmp_path): + """The eager streaming pre-check must record the miss in + last_restore_succeeded before raising — including overwriting a stale True + from a previous successful restore.""" + db_path = os.path.join(tmp_path, "test_flows.db") + persistence = SQLiteFlowPersistence(db_path) + + class EagerSignalFlow(Flow[TestState]): + @start() + @persist(persistence) + def step(self): + self.state.counter += 1 + + flow1 = EagerSignalFlow(persistence=persistence) + flow1.kickoff() + source_uuid = flow1.state.id + + # Prime the signal with a successful restore first. + flow2 = EagerSignalFlow(persistence=persistence) + flow2.kickoff(restore_from_state_id=source_uuid) + assert flow2.last_restore_succeeded is True + + # Eager streaming miss on the same instance: signal flips to False, not stale True. + with pytest.raises(ValueError): + flow2.stream_events( + restore_from_state_id="no-such-uuid", + raise_on_missing_state=True, + ) + assert flow2.last_restore_succeeded is False + + # Fresh instance failing in the astream pre-check: False, not None. + flow3 = EagerSignalFlow(persistence=persistence) + with pytest.raises(ValueError): + flow3.astream( + restore_from_state_id="no-such-uuid", + raise_on_missing_state=True, + ) + assert flow3.last_restore_succeeded is False + + # No-persistence eager failure records the miss too. + class NoPersistenceStreamFlow(Flow[TestState]): + @start() + def step(self): + self.state.counter += 1 + + flow4 = NoPersistenceStreamFlow() + with pytest.raises(ValueError): + flow4.stream_events( + restore_from_state_id="some-uuid", + raise_on_missing_state=True, + ) + assert flow4.last_restore_succeeded is False + + +def test_stream_conflict_check_wins_over_strict_restore_miss(tmp_path): + """stream_events/astream with both from_checkpoint and restore_from_state_id + raise the mutual-exclusion ValueError, not the restore-miss one, even with + raise_on_missing_state=True and a missing id (matches non-streaming precedence).""" + from crewai.state import CheckpointConfig + + db_path = os.path.join(tmp_path, "test_flows.db") + persistence = SQLiteFlowPersistence(db_path) + + class StreamConflictFlow(Flow[TestState]): + @start() + @persist(persistence) + def step(self): + self.state.counter += 1 + + flow = StreamConflictFlow(persistence=persistence) + with pytest.raises(ValueError) as excinfo: + flow.stream_events( + from_checkpoint=CheckpointConfig(), + restore_from_state_id="no-such-uuid", + raise_on_missing_state=True, + ) + msg = str(excinfo.value) + assert "Cannot combine" in msg + assert "from_checkpoint" in msg + + flow2 = StreamConflictFlow(persistence=persistence) + with pytest.raises(ValueError) as excinfo2: + flow2.astream( + from_checkpoint=CheckpointConfig(), + restore_from_state_id="no-such-uuid", + raise_on_missing_state=True, + ) + assert "Cannot combine" in str(excinfo2.value) + + def test_last_restore_succeeded_signal(tmp_path): """last_restore_succeeded is None when no restore was requested, False after a silent miss, and True after a successful restore."""