Address streaming contract review feedback

lib/crewai/src/crewai/events/event_bus.py

@@ -777,9 +777,7 @@ class CrewAIEventsBus:
             source: The object emitting the event
             event: The event instance to emit
         """
-        self._register_source(source)
-        event.emission_sequence = get_next_emission_sequence()
-        self._record_event(event)
+        self._prepare_event(source, event)
 
         event_type = type(event)
 

lib/crewai/src/crewai/experimental/conversational_mixin.py

@@ -45,6 +45,7 @@ from crewai.experimental.conversational import (
     _conversational_only,
     message_to_llm_dict,
 )
+from crewai.flow.async_feedback import HumanFeedbackPending
 from crewai.flow.conversation import (
     append_message as _append_conversation_message,
     get_conversation_messages,
@@ -382,7 +383,6 @@ class _ConversationalMixin:
             self._pending_intents = list(intents) if intents else None
             self._pending_intent_llm = intent_llm
 
-            failed_event: ConversationTurnFailedEvent | None = None
             try:
                 if "from_checkpoint" not in kickoff_kwargs:
                     self._reset_turn_execution_state()
@@ -407,6 +407,8 @@ class _ConversationalMixin:
                     and self._is_public_turn_result(result)
                 ):
                     self.append_assistant_message(self._stringify_result(result))
+            except HumanFeedbackPending as exc:
+                return exc
             except Exception as exc:
                 failed_event = ConversationTurnFailedEvent(
                     type="conversation_turn_failed",

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

@@ -1950,6 +1950,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             return self.stream_events(
                 inputs=inputs,
                 input_files=input_files,
+                from_checkpoint=from_checkpoint,
                 restore_from_state_id=restore_from_state_id,
             )
 
@@ -2015,6 +2016,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             return self.astream(
                 inputs=inputs,
                 input_files=input_files,
+                from_checkpoint=from_checkpoint,
                 restore_from_state_id=restore_from_state_id,
             )
 

lib/crewai/src/crewai/llms/base_llm.py

@@ -339,20 +339,16 @@ class BaseLLM(BaseModel, ABC):
         output_holder: list[StreamSession[Any]] = []
 
         def run_llm_call() -> Any:
-            original_stream = self.stream
-            try:
-                self.stream = True
-                return self.call(
-                    messages=messages,
-                    tools=tools,
-                    callbacks=callbacks,
-                    available_functions=available_functions,
-                    from_task=from_task,
-                    from_agent=from_agent,
-                    response_model=response_model,
-                )
-            finally:
-                self.stream = original_stream
+            streaming_llm = self.model_copy(update={"stream": True})
+            return streaming_llm.call(
+                messages=messages,
+                tools=tools,
+                callbacks=callbacks,
+                available_functions=available_functions,
+                from_task=from_task,
+                from_agent=from_agent,
+                response_model=response_model,
+            )
 
         stream_session: StreamSession[Any] = StreamSession(
             sync_iterator=create_frame_generator(state, run_llm_call, output_holder)

lib/crewai/tests/test_flow_conversation.py

@@ -29,6 +29,7 @@ from crewai.experimental import (
     RouterConfig,
 )
 from crewai.flow import Flow, ChatState, listen, start
+from crewai.flow.async_feedback import HumanFeedbackPending, PendingFeedbackContext
 from crewai.flow.flow_context import (
     current_flow_defer_trace_finalization,
     current_flow_id,
@@ -220,6 +221,28 @@ class TestConversationalFlow:
             if frame.type == "llm_stream_chunk"
         ] == ["po", "ng"]
 
+    def test_stream_turn_returns_pending_feedback_without_failure_event(self) -> None:
+        flow = ConversationalFlow()
+        pending = HumanFeedbackPending(
+            context=PendingFeedbackContext(
+                flow_id="session-1",
+                flow_class="tests.PendingFeedbackFlow",
+                method_name="review",
+                method_output="draft",
+                message="Please review",
+            )
+        )
+
+        def kickoff_side_effect(*_: Any, **__: Any) -> None:
+            raise pending
+
+        with patch.object(flow, "kickoff", side_effect=kickoff_side_effect):
+            stream = flow.stream_turn("review this", session_id="session-1")
+            frames = list(stream.events)
+
+        assert stream.result is pending
+        assert [frame.type for frame in frames] == ["conversation_turn_started"]
+
     def test_deferred_multi_turn_emits_single_flow_finished(self) -> None:
         """A deferred multi-turn session lands as one trace: exactly one
         ``FlowFinishedEvent`` is emitted at ``finalize_session_traces()``, not

lib/crewai/tests/test_stream_frames.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, ClassVar
 
 import pytest
 
@@ -59,7 +59,12 @@ class FrameFlow(Flow):
 
 
 class DirectStreamingLLM(BaseLLM):
+    call_stream_values: ClassVar[list[bool | None]] = []
+    call_instance_ids: ClassVar[list[int]] = []
+
     def call(self, messages: Any, *args: Any, **kwargs: Any) -> str:
+        self.call_stream_values.append(self.stream)
+        self.call_instance_ids.append(id(self))
         crewai_event_bus.emit(
             self,
             LLMStreamChunkEvent(
@@ -149,6 +154,8 @@ def test_flow_streaming_returns_iterable_frame_session() -> None:
 
 
 def test_direct_llm_stream_events_scope_and_restore_stream_flag() -> None:
+    DirectStreamingLLM.call_stream_values = []
+    DirectStreamingLLM.call_instance_ids = []
     llm = DirectStreamingLLM(model="gpt-4o-mini", stream=False)
 
     with llm.stream_events("hello") as stream:
@@ -158,6 +165,8 @@ def test_direct_llm_stream_events_scope_and_restore_stream_flag() -> None:
     assert frames[0].event["chunk"] == "hel"
     assert stream.result == "hello"
     assert llm.stream is False
+    assert DirectStreamingLLM.call_stream_values == [True]
+    assert DirectStreamingLLM.call_instance_ids != [id(llm)]
 
 
 @pytest.mark.asyncio

lib/crewai/tests/test_streaming.py

@@ -11,6 +11,7 @@ from crewai import Agent, Crew, Task
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.llm_events import LLMStreamChunkEvent, ToolCall, FunctionCall
 from crewai.flow.flow import Flow, start
+from crewai.state.checkpoint_config import CheckpointConfig
 from crewai.types.streaming import (
     AsyncStreamSession,
     CrewStreamingOutput,
@@ -509,6 +510,25 @@ class TestFlowKickoffStreaming:
         result = streaming.result
         assert result == "flow result"
 
+    def test_streaming_kickoff_passes_checkpoint_config_to_stream_events(self) -> None:
+        """stream=True preserves checkpoint config when routing to stream_events."""
+
+        class TestFlow(Flow[dict[str, Any]]):
+            @start()
+            def generate(self) -> str:
+                return "flow result"
+
+        flow = TestFlow()
+        flow.stream = True
+        checkpoint = CheckpointConfig()
+
+        with patch.object(flow, "stream_events", wraps=flow.stream_events) as spy:
+            streaming = flow.kickoff(from_checkpoint=checkpoint)
+            list(streaming)
+
+        assert spy.call_args.kwargs["from_checkpoint"] is checkpoint
+        assert streaming.result == "flow result"
+
 
 class TestFlowKickoffStreamingAsync:
     """Tests for Flow(stream=True).kickoff_async() method."""
@@ -611,6 +631,29 @@ class TestFlowKickoffStreamingAsync:
         result = streaming.result
         assert result == "async flow result"
 
+    @pytest.mark.asyncio
+    async def test_streaming_kickoff_async_passes_checkpoint_config_to_astream(
+        self,
+    ) -> None:
+        """stream=True preserves checkpoint config when routing to astream."""
+
+        class TestFlow(Flow[dict[str, Any]]):
+            @start()
+            async def generate(self) -> str:
+                return "async flow result"
+
+        flow = TestFlow()
+        flow.stream = True
+        checkpoint = CheckpointConfig()
+
+        with patch.object(flow, "astream", wraps=flow.astream) as spy:
+            streaming = await flow.kickoff_async(from_checkpoint=checkpoint)
+            async for _ in streaming:
+                pass
+
+        assert spy.call_args.kwargs["from_checkpoint"] is checkpoint
+        assert streaming.result == "async flow result"
+
 
 class TestStreamingEdgeCases:
     """Tests for edge cases in streaming functionality."""

lib/crewai/tests/test_streaming_integration.py

@@ -3,6 +3,8 @@
 import pytest
 
 from crewai import Agent, Crew, Task
+from crewai.events.event_bus import crewai_event_bus
+from crewai.events.types.llm_events import LLMStreamChunkEvent
 from crewai.flow.flow import Flow, start
 from crewai.types.streaming import AsyncStreamSession, CrewStreamingOutput, StreamSession
 
@@ -232,6 +234,14 @@ class TestStreamingFlowIntegration:
 
             @start()
             def execute(self) -> str:
+                crewai_event_bus.emit(
+                    self,
+                    LLMStreamChunkEvent(
+                        type="llm_stream_chunk",
+                        chunk="Flow result",
+                        call_id="call-1",
+                    ),
+                )
                 return "Flow result"
 
         flow = SimpleFlow()
@@ -241,8 +251,10 @@ class TestStreamingFlowIntegration:
             pass
 
         assert streaming.is_completed is True
-        full_text = "".join(frame.content for frame in streaming.frames)
-        assert isinstance(full_text, str)
+        content_frames = [frame for frame in streaming.frames if frame.content]
+        full_text = "".join(frame.content for frame in content_frames)
+        assert full_text == "Flow result"
+        assert len(content_frames) == 1
         assert len(streaming.frames) > 0
 
         result = streaming.result

lib/crewai/tests/utilities/events/test_async_event_bus.py

@@ -9,6 +9,7 @@ import pytest
 
 from crewai.events.base_events import BaseEvent
 from crewai.events.event_bus import crewai_event_bus
+from crewai.events.stream_context import add_stream_sink, reset_stream_sinks
 
 
 class AsyncTestEvent(BaseEvent):
@@ -53,6 +54,24 @@ async def test_aemit_with_async_handlers():
         assert received_events[0] == event
 
 
+@pytest.mark.asyncio
+async def test_aemit_publishes_to_active_stream_sinks():
+    published_events = []
+
+    def sink(source: object, event: BaseEvent) -> None:
+        published_events.append((source, event))
+
+    event = AsyncTestEvent(type="async_test")
+    token = add_stream_sink(sink)
+    try:
+        await crewai_event_bus.aemit("test_source", event)
+    finally:
+        reset_stream_sinks(token)
+
+    assert published_events == [("test_source", event)]
+    assert event.emission_sequence is not None
+
+
 @pytest.mark.asyncio
 async def test_multiple_async_handlers():
     received_events_1 = []