[EPD-196] Reset restore signal eagerly when a streaming session is created

- stream_events/astream (and thus kickoff/kickoff_async with stream=True)
  now reset last_restore_succeeded to None before returning the session,
  matching kickoff_async's reset-at-start semantics. Previously the
  property held the prior kickoff's stale True/False until frames were
  consumed (Bugbot round 3).
- The eager strict-raise path is unchanged: with raise_on_missing_state a
  miss still sets the signal to False before raising at session creation.
- Document streaming timing on the last_restore_succeeded property.
- Failing-first sync + async tests: stale True from a prior restore reads
  None right after obtaining a new session, then False after consumption.

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

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

@@ -1670,6 +1670,14 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         the lookup missed (or no persistence backend was configured) and the
         run proceeded without hydration. Lets callers detect silent restore
         misses without parsing logs.
+
+        Streaming timing: streaming kickoffs (``stream_events``/``astream``/
+        ``kickoff(stream=True)``) defer execution until frames are consumed.
+        Obtaining the session resets this signal to ``None``; it transitions
+        to ``True``/``False`` only once the deferred run evaluates the
+        restore, so read it after consuming the stream (with
+        ``raise_on_missing_state=True``, a miss still fails eagerly at
+        session creation and sets this to ``False``).
         """
         return self._last_restore_succeeded
 
@@ -1929,6 +1937,10 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         returned, so callers do not have to consume frames to see the error.
         """
         self._check_restore_conflict(from_checkpoint, restore_from_state_id)
+        # The deferred run only evaluates restore_from_state_id once frames are
+        # consumed; reset the signal now so a caller reading it right after
+        # session creation sees None (no outcome yet) instead of a stale value.
+        self._last_restore_succeeded = None
         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)
@@ -1971,6 +1983,10 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         returned, so callers do not have to consume frames to see the error.
         """
         self._check_restore_conflict(from_checkpoint, restore_from_state_id)
+        # The deferred run only evaluates restore_from_state_id once frames are
+        # consumed; reset the signal now so a caller reading it right after
+        # session creation sees None (no outcome yet) instead of a stale value.
+        self._last_restore_succeeded = None
         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)

lib/crewai/tests/test_flow_persistence.py

@@ -567,6 +567,68 @@ def test_stream_eager_strict_miss_sets_last_restore_succeeded_false(tmp_path):
     assert flow4.last_restore_succeeded is False
 
 
+def test_stream_session_creation_resets_last_restore_succeeded(tmp_path):
+    """Obtaining a streaming session resets last_restore_succeeded to None (no
+    outcome yet) instead of exposing the previous kickoff's stale value; the
+    signal then transitions to the new outcome as frames are consumed."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class StreamResetFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = StreamResetFlow(persistence=persistence)
+    flow1.kickoff()
+    source_uuid = flow1.state.id
+
+    # Prime a stale True from a successful non-streaming restore.
+    flow2 = StreamResetFlow(persistence=persistence)
+    flow2.kickoff(restore_from_state_id=source_uuid)
+    assert flow2.last_restore_succeeded is True
+
+    # New streaming kickoff with a different (missing) restore id: the deferred
+    # run has not evaluated the restore yet, so the signal must read None, not
+    # the stale True.
+    session = flow2.stream_events(restore_from_state_id="no-such-uuid")
+    assert flow2.last_restore_succeeded is None
+
+    with session:
+        list(session.events)
+    assert flow2.last_restore_succeeded is False
+
+
+@pytest.mark.asyncio
+async def test_astream_session_creation_resets_last_restore_succeeded(tmp_path):
+    """astream session creation resets the restore signal to None before the
+    deferred run evaluates restore_from_state_id."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class AsyncStreamResetFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = AsyncStreamResetFlow(persistence=persistence)
+    await flow1.kickoff_async()
+    source_uuid = flow1.state.id
+
+    flow2 = AsyncStreamResetFlow(persistence=persistence)
+    await flow2.kickoff_async(restore_from_state_id=source_uuid)
+    assert flow2.last_restore_succeeded is True
+
+    stream = flow2.astream(restore_from_state_id="no-such-uuid")
+    assert flow2.last_restore_succeeded is None
+
+    async with stream:
+        _ = [frame async for frame in stream.events]
+    assert flow2.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