[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>

docs/edge/en/guides/flows/mastering-flow-state.mdx

@@ -385,7 +385,7 @@ flow_2.kickoff(restore_from_state_id=flow_1.state.id)
 
 Behavior notes:
 
-- `restore_from_state_id` not found in persistence → the kickoff falls back silently to default behavior (mirrors the existing `inputs["id"]` resume not-found behavior). No exception is raised.
+- `restore_from_state_id` not found in persistence → the kickoff falls back silently to default behavior (mirrors the existing `inputs["id"]` resume not-found behavior). No exception is raised by default. Pass `raise_on_missing_state=True` to get a `ValueError` instead of a silent fresh run, or check `flow.last_restore_succeeded` after kickoff (`None` = no restore requested, `True` = hydrated, `False` = miss).
 - Combining `restore_from_state_id` with `from_checkpoint` raises a `ValueError` — they target different state systems (`@persist` vs. Checkpointing) and cannot be combined.
 - `restore_from_state_id=None` (default) is byte-identical to a kickoff without the parameter.
 - Pinning `inputs["id"]` while forking means the new run shares a persistence key with another flow — usually you want only `restore_from_state_id`.

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

@@ -730,6 +730,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
     _usage_aggregation_handler: Callable[..., Any] | None = PrivateAttr(default=None)
     _persist_backends: dict[int, FlowPersistence] = PrivateAttr(default_factory=dict)
     _instance_persistence: bool = PrivateAttr(default=False)
+    _last_restore_succeeded: bool | None = PrivateAttr(default=None)
 
     def __class_getitem__(cls: type[Flow[T]], item: type[T]) -> type[Flow[T]]:  # type: ignore[override]
         class _FlowGeneric(cls):  # type: ignore[valid-type,misc]
@@ -1660,6 +1661,26 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
     def state(self) -> T:
         return cast(T, self._state)
 
+    @property
+    def last_restore_succeeded(self) -> bool | None:
+        """Outcome of the most recent ``restore_from_state_id`` request.
+
+        ``None`` when the latest kickoff did not request a restore, ``True``
+        when the referenced state was found and hydrated, and ``False`` when
+        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
+
     @property
     def method_outputs(self) -> list[Any]:
         """Returns the list of all outputs from executed methods."""
@@ -1851,14 +1872,76 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 if hasattr(self._state, key):
                     object.__setattr__(self._state, key, value)
 
+    @staticmethod
+    def _missing_state_error(restore_from_state_id: str) -> ValueError:
+        return ValueError(
+            f"No flow state found for restore_from_state_id: '{restore_from_state_id}'"
+        )
+
+    @staticmethod
+    def _no_persistence_error(restore_from_state_id: str) -> ValueError:
+        return ValueError(
+            "Cannot restore from restore_from_state_id: "
+            f"'{restore_from_state_id}'; no persistence backend "
+            "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,
+        raise_on_missing_state: bool,
+    ) -> None:
+        """Eagerly validate a strict restore request before deferred execution.
+
+        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. 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(
         self,
         inputs: dict[str, Any] | None = None,
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
         restore_from_state_id: str | None = None,
+        raise_on_missing_state: bool = False,
     ) -> StreamSession[Any]:
-        """Run the flow and stream all scoped public ``StreamFrame`` events."""
+        """Run the flow and stream all scoped public ``StreamFrame`` events.
+
+        With ``raise_on_missing_state=True``, a ``restore_from_state_id``
+        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)
+        # 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)
         output_holder: list[StreamSession[Any]] = []
@@ -1872,6 +1955,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     input_files=input_files,
                     from_checkpoint=from_checkpoint,
                     restore_from_state_id=restore_from_state_id,
+                    raise_on_missing_state=raise_on_missing_state,
                 )
             except HumanFeedbackPending as e:
                 return e
@@ -1890,8 +1974,20 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
         restore_from_state_id: str | None = None,
+        raise_on_missing_state: bool = False,
     ) -> AsyncStreamSession[Any]:
-        """Run the flow asynchronously and stream scoped public frames."""
+        """Run the flow asynchronously and stream scoped public frames.
+
+        With ``raise_on_missing_state=True``, a ``restore_from_state_id``
+        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)
+        # 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)
         output_holder: list[AsyncStreamSession[Any]] = []
@@ -1905,6 +2001,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     input_files=input_files,
                     from_checkpoint=from_checkpoint,
                     restore_from_state_id=restore_from_state_id,
+                    raise_on_missing_state=raise_on_missing_state,
                 )
             except HumanFeedbackPending as e:
                 return e
@@ -1923,6 +2020,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
         restore_from_state_id: str | None = None,
+        raise_on_missing_state: bool = False,
     ) -> Any | StreamSession[Any]:
         """Start the flow execution in a synchronous context.
 
@@ -1940,18 +2038,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 pinned), so its ``@persist`` writes land under a separate
                 persistence key and the source flow's history is preserved.
                 If the referenced state is not found, the kickoff falls back
-                silently to baseline behavior. Cannot be combined with
-                ``from_checkpoint``; passing both raises ``ValueError``.
+                silently to baseline behavior (see ``raise_on_missing_state``
+                to opt out). Cannot be combined with ``from_checkpoint``;
+                passing both raises ``ValueError``.
+            raise_on_missing_state: When True and ``restore_from_state_id`` is
+                provided but the referenced state cannot be loaded, raise
+                ``ValueError`` instead of silently running without hydration.
+                Defaults to False. ``last_restore_succeeded`` reports the
+                restore outcome either way.
 
         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)
@@ -1961,6 +2060,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 input_files=input_files,
                 from_checkpoint=from_checkpoint,
                 restore_from_state_id=restore_from_state_id,
+                raise_on_missing_state=raise_on_missing_state,
             )
 
         async def _run_flow() -> Any:
@@ -1968,6 +2068,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 inputs,
                 input_files,
                 restore_from_state_id=restore_from_state_id,
+                raise_on_missing_state=raise_on_missing_state,
             )
 
         runtime_scope = crewai_event_bus._enter_runtime_scope()
@@ -1988,6 +2089,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
         restore_from_state_id: str | None = None,
+        raise_on_missing_state: bool = False,
     ) -> Any | AsyncStreamSession[Any]:
         """Start the flow execution asynchronously.
 
@@ -2006,18 +2108,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 run is assigned a fresh ``state.id`` (or ``inputs["id"]`` if
                 pinned), so subsequent ``@persist`` writes land under a
                 separate persistence key. If the referenced state is not
-                found, falls back silently to baseline. Cannot be combined
+                found, falls back silently to baseline (see
+                ``raise_on_missing_state`` to opt out). Cannot be combined
                 with ``from_checkpoint``; passing both raises ``ValueError``.
+            raise_on_missing_state: When True and ``restore_from_state_id`` is
+                provided but the referenced state cannot be loaded, raise
+                ``ValueError`` instead of silently running without hydration.
+                Defaults to False. ``last_restore_succeeded`` reports the
+                restore outcome either way.
 
         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)
@@ -2027,6 +2130,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 input_files=input_files,
                 from_checkpoint=from_checkpoint,
                 restore_from_state_id=restore_from_state_id,
+                raise_on_missing_state=raise_on_missing_state,
             )
 
         ctx = baggage.set_baggage("flow_inputs", inputs or {})
@@ -2090,7 +2194,10 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             # available, hydrate self._state from the source UUID's latest snapshot
             # and reassign state.id to a fresh value so subsequent @persist writes
             # don't extend the source flow's history. If the source state is not
-            # found, fall through silently to the existing inputs handling.
+            # found, fall through silently to the existing inputs handling
+            # (unless raise_on_missing_state is set). last_restore_succeeded
+            # records the outcome for callers either way.
+            self._last_restore_succeeded = None
             fork_succeeded = False
             if restore_from_state_id is not None and self.persistence is not None:
                 stored_state = self.persistence.load_state(restore_from_state_id)
@@ -2111,12 +2218,20 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     elif isinstance(self._state, BaseModel):
                         setattr(self._state, "id", new_state_id)  # noqa: B010
                     fork_succeeded = True
+                    self._last_restore_succeeded = True
                 else:
+                    self._last_restore_succeeded = False
+                    if raise_on_missing_state:
+                        raise self._missing_state_error(restore_from_state_id)
                     self._log_flow_event(
                         "No flow state found for restore_from_state_id: "
                         f"{restore_from_state_id}; proceeding without hydration",
                         color="yellow",
                     )
+            elif restore_from_state_id is not None:
+                self._last_restore_succeeded = False
+                if raise_on_missing_state:
+                    raise self._no_persistence_error(restore_from_state_id)
 
             if inputs:
                 # Override the id in the state if it exists in inputs.
@@ -2379,6 +2494,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
         restore_from_state_id: str | None = None,
+        raise_on_missing_state: bool = False,
     ) -> Any | AsyncStreamSession[Any]:
         """Native async method to start the flow execution. Alias for kickoff_async.
 
@@ -2390,6 +2506,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             restore_from_state_id: Optional UUID of a previously-persisted flow
                 whose latest snapshot should hydrate this run's state. See
                 ``kickoff_async`` for full semantics.
+            raise_on_missing_state: When True, a ``restore_from_state_id``
+                lookup miss raises ``ValueError`` instead of silently running
+                without hydration. See ``kickoff_async``.
 
         Returns:
             The final output from the flow, which is the result of the last executed method.
@@ -2399,6 +2518,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             input_files,
             from_checkpoint,
             restore_from_state_id=restore_from_state_id,
+            raise_on_missing_state=raise_on_missing_state,
         )
 
     async def _replay_recorded_events(self) -> None:

lib/crewai/tests/llms/openai/test_openai.py

@@ -212,6 +212,18 @@ def test_openai_completion_module_is_imported(monkeypatch):
     """
     module_name = "crewai.llms.providers.openai.completion"
 
+    # The LLM(...) call below re-imports the module, which also rebinds the
+    # parent package's `completion` attribute to the NEW module object.
+    # monkeypatch.delitem only restores the sys.modules entry at teardown, so
+    # without also restoring the package attribute the two diverge for the
+    # rest of the process. On Python 3.10 `from ...openai.completion import X`
+    # then resolves through the stale package attribute while mock.patch
+    # targets the restored sys.modules entry, making later patches of
+    # OpenAICompletion silently miss (breaks e.g. test_azure_responses.py).
+    import crewai.llms.providers.openai as openai_pkg
+
+    if hasattr(openai_pkg, "completion"):
+        monkeypatch.setattr(openai_pkg, "completion", openai_pkg.completion)
     monkeypatch.delitem(sys.modules, module_name, raising=False)
 
     LLM(model="gpt-4o")

lib/crewai/tests/test_flow_persistence.py

@@ -331,6 +331,376 @@ def test_restore_from_state_id_not_found_silent_fallback(tmp_path):
     assert flow.state.id != "no-such-uuid"
 
 
+def test_restore_from_state_id_not_found_raises_when_strict(tmp_path):
+    """raise_on_missing_state=True turns a restore_from_state_id lookup miss into a
+    ValueError instead of a silent fresh run."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class StrictFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow = StrictFlow(persistence=persistence)
+    with pytest.raises(ValueError) as excinfo:
+        flow.kickoff(
+            restore_from_state_id="no-such-uuid",
+            raise_on_missing_state=True,
+        )
+
+    msg = str(excinfo.value)
+    assert "restore_from_state_id" in msg
+    assert "no-such-uuid" in msg
+    # The flow never ran: state is untouched and the outcome signal reports the miss.
+    assert flow.state.counter == 0
+    assert flow.last_restore_succeeded is False
+
+
+def test_restore_from_state_id_strict_with_valid_id_restores(tmp_path):
+    """raise_on_missing_state=True does not change behavior when the source state
+    exists: hydrate from the snapshot, mint a fresh state.id, report success."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class StrictForkFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = StrictForkFlow(persistence=persistence)
+    flow1.kickoff()
+    source_uuid = flow1.state.id
+    assert flow1.state.counter == 1
+
+    flow2 = StrictForkFlow(persistence=persistence)
+    flow2.kickoff(
+        restore_from_state_id=source_uuid,
+        raise_on_missing_state=True,
+    )
+
+    assert flow2.state.id != source_uuid
+    assert flow2.state.counter == 2
+    assert flow2.last_restore_succeeded is True
+
+
+def test_strict_restore_without_persistence_raises():
+    """raise_on_missing_state=True with restore_from_state_id but no persistence
+    backend raises instead of silently skipping hydration."""
+
+    class NoPersistenceFlow(Flow[TestState]):
+        @start()
+        def step(self):
+            self.state.counter += 1
+
+    flow = NoPersistenceFlow()
+    with pytest.raises(ValueError) as excinfo:
+        flow.kickoff(
+            restore_from_state_id="some-uuid",
+            raise_on_missing_state=True,
+        )
+    msg = str(excinfo.value)
+    assert "restore_from_state_id" in msg
+    assert "persistence" in msg
+    assert flow.last_restore_succeeded is False
+
+
+def test_stream_events_strict_restore_miss_raises_eagerly(tmp_path):
+    """stream_events with raise_on_missing_state=True raises at session creation
+    time on a restore_from_state_id miss, not while consuming frames."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class StreamStrictFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow = StreamStrictFlow(persistence=persistence)
+    with pytest.raises(ValueError) as excinfo:
+        flow.stream_events(
+            restore_from_state_id="no-such-uuid",
+            raise_on_missing_state=True,
+        )
+    assert "no-such-uuid" in str(excinfo.value)
+    # Nothing ran.
+    assert flow.state.counter == 0
+
+
+def test_kickoff_stream_strict_restore_miss_raises_eagerly(tmp_path):
+    """kickoff with stream=True and raise_on_missing_state=True raises immediately
+    on a restore miss instead of returning a StreamSession."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class StreamingKickoffFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow = StreamingKickoffFlow(persistence=persistence)
+    flow.stream = True
+    with pytest.raises(ValueError) as excinfo:
+        flow.kickoff(
+            restore_from_state_id="no-such-uuid",
+            raise_on_missing_state=True,
+        )
+    assert "no-such-uuid" in str(excinfo.value)
+    assert flow.state.counter == 0
+
+
+def test_stream_events_strict_restore_valid_id_streams(tmp_path):
+    """stream_events with raise_on_missing_state=True streams normally when the
+    source state exists: hydrates, mints a fresh state.id, reports success."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class StreamForkFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = StreamForkFlow(persistence=persistence)
+    flow1.kickoff()
+    source_uuid = flow1.state.id
+
+    flow2 = StreamForkFlow(persistence=persistence)
+    with flow2.stream_events(
+        restore_from_state_id=source_uuid,
+        raise_on_missing_state=True,
+    ) as stream:
+        list(stream.events)
+
+    assert flow2.state.id != source_uuid
+    assert flow2.state.counter == 2
+    assert flow2.last_restore_succeeded is True
+
+
+@pytest.mark.asyncio
+async def test_astream_strict_restore_miss_raises_eagerly(tmp_path):
+    """astream and kickoff_async(stream=True) raise at call time on a restore miss
+    with raise_on_missing_state=True."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class AsyncStreamStrictFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow = AsyncStreamStrictFlow(persistence=persistence)
+    with pytest.raises(ValueError) as excinfo:
+        flow.astream(
+            restore_from_state_id="no-such-uuid",
+            raise_on_missing_state=True,
+        )
+    assert "no-such-uuid" in str(excinfo.value)
+    assert flow.state.counter == 0
+
+    flow2 = AsyncStreamStrictFlow(persistence=persistence)
+    flow2.stream = True
+    with pytest.raises(ValueError):
+        await flow2.kickoff_async(
+            restore_from_state_id="no-such-uuid",
+            raise_on_missing_state=True,
+        )
+    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_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
+    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."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class SignalFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = SignalFlow(persistence=persistence)
+    assert flow1.last_restore_succeeded is None
+    flow1.kickoff()
+    # No restore requested: signal stays None.
+    assert flow1.last_restore_succeeded is None
+    source_uuid = flow1.state.id
+
+    # Silent miss (default non-strict): the run proceeds but the signal is False.
+    flow2 = SignalFlow(persistence=persistence)
+    flow2.kickoff(restore_from_state_id="no-such-uuid")
+    assert flow2.state.counter == 1
+    assert flow2.last_restore_succeeded is False
+
+    # Successful restore: the signal is True.
+    flow3 = SignalFlow(persistence=persistence)
+    flow3.kickoff(restore_from_state_id=source_uuid)
+    assert flow3.state.counter == 2
+    assert flow3.last_restore_succeeded is True
+
+    # A later kickoff without a restore request resets the signal to None.
+    flow3.kickoff()
+    assert flow3.last_restore_succeeded is None
+
+
 def test_restore_from_state_id_none_is_no_op(tmp_path):
     """restore_from_state_id=None (default) preserves baseline kickoff behavior."""
     db_path = os.path.join(tmp_path, "test_flows.db")
@@ -452,6 +822,65 @@ async def test_akickoff_pinned_fork(tmp_path):
     assert persistence.load_state(pinned_uuid)["counter"] == 2
 
 
+@pytest.mark.asyncio
+async def test_restore_from_state_id_not_found_raises_when_strict_async(tmp_path):
+    """kickoff_async and akickoff honor raise_on_missing_state on a lookup miss."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class AsyncStrictFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow = AsyncStrictFlow(persistence=persistence)
+    with pytest.raises(ValueError) as excinfo:
+        await flow.kickoff_async(
+            restore_from_state_id="no-such-uuid",
+            raise_on_missing_state=True,
+        )
+    assert "no-such-uuid" in str(excinfo.value)
+    assert flow.state.counter == 0
+    assert flow.last_restore_succeeded is False
+
+    flow_alias = AsyncStrictFlow(persistence=persistence)
+    with pytest.raises(ValueError):
+        await flow_alias.akickoff(
+            restore_from_state_id="no-such-uuid",
+            raise_on_missing_state=True,
+        )
+    assert flow_alias.last_restore_succeeded is False
+
+
+@pytest.mark.asyncio
+async def test_restore_from_state_id_strict_with_valid_id_restores_async(tmp_path):
+    """kickoff_async with raise_on_missing_state=True restores normally when the
+    source state exists and reports success."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+
+    class AsyncStrictForkFlow(Flow[TestState]):
+        @start()
+        @persist(persistence)
+        def step(self):
+            self.state.counter += 1
+
+    flow1 = AsyncStrictForkFlow(persistence=persistence)
+    await flow1.kickoff_async()
+    source_uuid = flow1.state.id
+
+    flow2 = AsyncStrictForkFlow(persistence=persistence)
+    await flow2.kickoff_async(
+        restore_from_state_id=source_uuid,
+        raise_on_missing_state=True,
+    )
+
+    assert flow2.state.id != source_uuid
+    assert flow2.state.counter == 2
+    assert flow2.last_restore_succeeded is True
+
+
 @pytest.mark.asyncio
 async def test_akickoff_fork_conflict_with_from_checkpoint_raises():
     """akickoff must raise the same conflict ValueError as kickoff/kickoff_async when