Repository: stanfordnlp/dspy

Issue:
[Bug] Status Streaming randomly doesn't work.
### What happened?

I took the code from here https://github.com/stanfordnlp/dspy/pull/7630 and tried to run it with a tool like 

```python
def write_invitation_letter(participant_name: str, event_info: CalendarEvent):
    sleep(5)
    return f"It's my honor to invite {participant_name} to event {event_info.name} on {event_info.date}"
```

Output this in the streaming of status

```
[0.00s] I am calling ReAct!
[0.00s] I am just calling the LLM!
[3.55s] I have finished calling the LLM!
[8.56s] Calling tool write_invitation_letter...
[8.56s] Tool calling finished! Querying the LLM with tool calling results...
```

As you can see the yield of the "calling tool" is yielded only when the tool calling finished.

I tried running it synchronously, async... it fails in all cases.

### Steps to reproduce

**CODE WITH ISSUE:**

```python
import asyncio
import time
from typing import Any

from pydantic import BaseModel

import dspy
from dspy.streaming import StatusMessage, StatusMessageProvider

lm = dspy.LM("openai/gpt-4o-mini")
dspy.settings.configure(lm=lm)


class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]


def write_invitation_letter(participant_name: str, event_info: CalendarEvent):
    time.sleep(5)
    if participant_name not in event_info.participants:
        return None
    return f"It's my honor to invite {participant_name} to event {event_info.name} on {event_info.date}"


class InvitationSignature(dspy.Signature):
    participant_name: str = dspy.InputField(desc="The name of the participant to invite")
    event_info: CalendarEvent = dspy.InputField(desc="The information about the event")
    invitation_letter: str = dspy.OutputField(desc="The invitation letter to be sent to the participant")


class CustomStatusMessageProvider(StatusMessageProvider):
    def lm_start_status_message(self, instance: Any, inputs: Any):
        return "I am just calling the LLM!"

    def lm_end_status_message(self, outputs: Any):
        return "I have finished calling the LLM!"

    def module_start_status_message(self, instance: Any, inputs: Any):
        if isinstance(instance, dspy.ReAct):
            return "I am calling ReAct!"


react = dspy.ReAct(InvitationSignature, tools=[write_invitation_letter])
stream_react = dspy.streamify(react, status_message_provider=CustomStatusMessageProvider())


async def main():
    output = stream_react(
        participant_name="Alice",
        event_info=CalendarEvent(
            name="Science Fair",
            date="Friday",
            participants=["Alice", "Bob"],
        ),
    )
    return_value = None
    start_time = time.time()
    async for value in output:
        if isinstance(value, StatusMessage):
            elapsed = time.time() - start_time
            print(f"[{elapsed:.2f}s] {value.message}")
        else:
            return_value = value
    return return_value


print(asyncio.run(main()))
```
Output (it didn't even stream anything):
```
[5.03s] I am calling ReAct!
[5.03s] I am just calling the LLM!
[5.03s] I have finished calling the LLM!
[5.03s] Calling tool write_invitation_letter...
[5.03s] Tool calling finished! Querying the LLM with tool calling results...
[5.03s] I am just calling the LLM!
[5.03s] I have finished calling the LLM!
[5.03s] I am just calling the LLM!
[5.03s] I have finished calling the LLM!
Prediction(
    trajectory={'thought_0': "I have the participant's name and the event information. I can now proceed to write the invitation letter for Alice to the Science Fair.", 'tool_name_0': 'write_invitation_letter', 'tool_args_0': {'participant_name': 'Alice', 'event_info': {'name': 'Science Fair', 'date': 'Friday', 'participants': ['Alice', 'Bob']}}, 'observation_0': "It's my honor to invite Alice to event Science Fair on Friday", 'thought_1': 'I have successfully written the invitation letter for Alice to the Science Fair. Since I have all the necessary information, I can now finish the task.', 'tool_name_1': 'finish', 'tool_args_1': {}, 'observation_1': 'Completed.'},
    reasoning='Alice is one of the participants invited to the Science Fair, which is scheduled for Friday. It is important to formally invite her to ensure she is aware of the event and can prepare accordingly.',
    invitation_letter='Dear Alice,\n\nI am pleased to invite you to the Science Fair taking place this Friday. Your participation is highly valued, and we look forward to seeing the innovative projects you will showcase.\n\nBest regards,\n\n[Your Name]'
)
```

### DSPy version

v2.6.27

Discussion:
@lambda-science: Same pattern in v3.0.0b1@lambda-science: Using bedrock & sync logic it's a bit better but still, tool calling status reporting happens at the same time:

```python
import asyncio
import os
import time
from typing import Any

from pydantic import BaseModel

import dspy
from dspy.streaming import StatusMessage, StatusMessageProvider

dspy.configure(
    lm=dspy.LM(
        provider="litellm",
        model="bedrock/us.anthropic.claude-3-5-haiku-20241022-v1:0",
        aws_region_name="us-east-1",
        cache=False,
    )
)

class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]


def write_invitation_letter(participant_name: str, event_info: CalendarEvent):
    time.sleep(5)
    if participant_name not in event_info.participants:
        return None
    return f"It's my honor to invite {participant_name} to event {event_info.name} on {event_info.date}"


class InvitationSignature(dspy.Signature):
    participant_name: str = dspy.InputField(desc="The name of the participant to invite")
    event_info: CalendarEvent = dspy.InputField(desc="The information about the event")
    invitation_letter: str = dspy.OutputField(desc="The invitation letter to be sent to the participant")


class CustomStatusMessageProvider(StatusMessageProvider):
    def lm_start_status_message(self, instance: Any, inputs: Any):
        return "I am just calling the LLM!"

    def lm_end_status_message(self, outputs: Any):
        return "I have finished calling the LLM!"

    def module_start_status_message(self, instance: Any, inputs: Any):
        if isinstance(instance, dspy.ReAct):
            return "I am calling ReAct!"


react = dspy.ReAct(InvitationSignature, tools=[write_invitation_letter])
stream_react = dspy.streamify(react, status_message_provider=CustomStatusMessageProvider(), async_streaming=False)


def main():
    output = stream_react(
        participant_name="Alice",
        event_info=CalendarEvent(
            name="Science Fair",
            date="Friday",
            participants=["Alice", "Bob"],
        ),
    )
    return_value = None
    start_time = time.time()
    for value in output:
        if isinstance(value, StatusMessage):
            elapsed = time.time() - start_time
            print(f"[{elapsed:.2f}s] {value.message}")
        else:
            return_value = value
    return return_value


print(main())
```

```
/Users/cmeyer/code/project-pythia/.venv/bin/python /Users/cmeyer/Library/Application Support/JetBrains/PyCharm2025.1/scratches/scratch_13.py 
[0.00s] I am calling ReAct!
[0.00s] I am just calling the LLM!
[3.67s] I have finished calling the LLM!
[8.69s] Calling tool write_invitation_letter...
[8.69s] Tool calling finished! Querying the LLM with tool calling results...
[8.69s] I am just calling the LLM!
[10.66s] I have finished calling the LLM!
[10.66s] I am just calling the LLM!
[18.05s] I have finished calling the LLM!
Prediction(
    trajectory={'thought_0': 'I will write an invitation letter for Alice to the Science Fair event. I have all the necessary information to complete this task, so I can directly use the write_invitation_letter tool.', 'tool_name_0': 'write_invitation_letter', 'tool_args_0': {'participant_name': 'Alice', 'event_info': {'name': 'Science Fair', 'date': 'Friday', 'participants': ['Alice', 'Bob']}}, 'observation_0': "It's my honor to invite Alice to event Science Fair on Friday", 'thought_1': 'The invitation letter has been successfully written for Alice to the Science Fair event. Since the task is complete, I will now use the finish tool to mark the task as completed.', 'tool_name_1': 'finish', 'tool_args_1': {}, 'observation_1': 'Completed.'},
    reasoning='I will craft an invitation letter for Alice to the Science Fair event. The event is scheduled for Friday, and Alice is one of the participants. The invitation should be formal, informative, and encouraging, highlighting the importance of the Science Fair and inviting her to participate.',
    invitation_letter='Dear Alice,\n\nI am pleased to invite you to the upcoming Science Fair, which will be held this Friday. As a valued participant, we are excited to have you join us for this engaging and innovative event.\n\nThe Science Fair provides an excellent opportunity to showcase scientific creativity, explore cutting-edge ideas, and connect with fellow science enthusiasts. Your participation is not only welcome but also crucial to the success of this event.\n\nPlease confirm your attendance and come prepared to share your passion for science. We look forward to seeing you there!\n\nWarm regards,\nEvent Organizing Committee'
)
```@lambda-science: Now if I put some print statement inside my `class CustomStatusMessageProvider(StatusMessageProvider` I see that at this point, it behaves like expected, it's called at the right time and return at the right time. So the problem is at the `yielding` phase ?

```python
class CustomStatusMessageProvider(StatusMessageProvider):
    def tool_start_status_message(self, instance, inputs):
        from datetime import datetime
        print(f"[{datetime.now().isoformat()}] status tool called !")
        return f"`Agent is using tool {instance.name}...`"

    def tool_end_status_message(self, outputs):
        from datetime import datetime
        print(f"[{datetime.now().isoformat()}] status tool finished !")
        return "`Finished using tool.`"
```
Output:
```
[2025-06-19T13:04:40.667019] status tool called !
[2025-06-19T13:04:45.676118] status tool finished !
[8.15s] `Agent is using tool write_invitation_letter...`
[8.15s] `Finished using tool.`
```
Print is at correct timing, yielding of status message is not@lambda-science: Same problem with exact code from tutorial documentation here https://dspy.ai/tutorials/streaming/#intermediate-status-streaming

```python
import time
import dspy
from dspy.streaming import StatusMessage
import asyncio

lm = dspy.LM("openai/gpt-4o-mini", cache=False)
dspy.settings.configure(lm=lm)

def double_the_number(x: int) -> int:
    """A simple tool that doubles the input number."""
    time.sleep(5)
    return 2 * x

class MyModule(dspy.Module):
    def __init__(self):
        super().__init__()

        self.tool = dspy.Tool(double_the_number)
        self.predict = dspy.ChainOfThought("num1, num2->sum")

    def forward(self, num, **kwargs):
        num2 = self.tool(x=num)
        return self.predict(num1=num, num2=num2)


class MyStatusMessageProvider(dspy.streaming.StatusMessageProvider):
    def tool_start_status_message(self, instance, inputs):
        return f"Calling Tool {instance.name} with inputs {inputs}..."

    def tool_end_status_message(self, instance, outputs):
        return f"Tool finished with output: {outputs}!"


predict = MyModule()
stream_listeners = [
    # dspy.ChainOfThought has a built-in output field called "reasoning".
    dspy.streaming.StreamListener(signature_field_name="reasoning"),
]
stream_predict = dspy.streamify(
    predict,
    stream_listeners=stream_listeners,
)

async def read_output_stream():
    output = stream_predict(num=3)

    return_value = None
    start_time = time.time()
    async for chunk in output:
        elapsed = time.time() - start_time
        if isinstance(chunk, dspy.streaming.StreamResponse):
            print(f"[{elapsed:.2f}s] {chunk}")
        elif isinstance(chunk, dspy.Prediction):
            return_value = chunk
        elif isinstance(chunk, dspy.streaming.StatusMessage):
            print(f"[{elapsed:.2f}s] {chunk}")
    return return_value


program_output = asyncio.run(read_output_stream())
print("Final output: ", program_output)
```

```
/Users/cmeyer/code/project-pythia/.venv/bin/python /Users/cmeyer/Library/Application Support/JetBrains/PyCharm2025.1/scratches/scratch_12.py 
[5.01s] StatusMessage(message='Calling tool double_the_number...')
[5.01s] StatusMessage(message='Tool calling finished! Querying the LLM with tool calling results...')
[5.70s] StreamResponse(predict_name='predict.predict', signature_field_name='reasoning', chunk='To')
[5.71s] StreamResponse(predict_name='predict.predict', signature_field_name='reasoning', chunk=' find')
[5.72s] StreamResponse(predict_name='predict.predict', signature_field_name='reasoning', chunk=' the')
....
```@chenmoneygithub: @lambda-science Thanks for reporting the issue! This is a valid bug, we are working on a fix. 

Solution patch:
diff --git a/dspy/streaming/messages.py b/dspy/streaming/messages.py
index 9a3cf74c..49023e8a 100644
--- a/dspy/streaming/messages.py
+++ b/dspy/streaming/messages.py
@@ -1,7 +1,10 @@
 import asyncio
+import concurrent.futures
 from dataclasses import dataclass
 from typing import Any, Dict, Optional
 
+from asyncer import syncify
+
 from dspy.dsp.utils.settings import settings
 from dspy.utils.callback import BaseCallback
 
@@ -21,36 +24,29 @@ class StatusMessage:
 
 
 def sync_send_to_stream(stream, message):
-    """Send message to stream in a sync context, regardless of whether the caller is async or not."""
-    # Try to get current event loop, create one if none exists
-    try:
-        loop = asyncio.get_event_loop()
-    except RuntimeError:
-        # "There is no current event loop in thread" error
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
+    """Send message to stream in a sync context, regardless of event loop state."""
 
-    # If we're in an async context
-    if loop.is_running():
-        # In an async context, we need to use an approach that doesn't block
-        # Create a new thread and run a new event loop there
-        import concurrent.futures
+    async def _send():
+        await stream.send(message)
 
-        def run_async_in_new_loop():
+    try:
+        asyncio.get_running_loop()
+
+        # If we're in an event loop, offload to a new thread with its own event loop
+        def run_in_new_loop():
             new_loop = asyncio.new_event_loop()
             asyncio.set_event_loop(new_loop)
             try:
-                return new_loop.run_until_complete(stream.send(message))
+                return new_loop.run_until_complete(_send())
             finally:
                 new_loop.close()
 
-        # Run the function in a separate thread
         with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
-            future = executor.submit(run_async_in_new_loop)
-            return future.result()  # This shouldn't hang now
-    else:
-        # We're in a sync context, use run_until_complete
-        return loop.run_until_complete(stream.send(message))
+            future = executor.submit(run_in_new_loop)
+            return future.result()
+    except RuntimeError:
+        # Not in an event loop, safe to use a new event loop in this thread
+        return syncify(_send)()
 
 
 class StatusMessageProvider:


Test patch:
diff --git a/tests/streaming/test_streaming.py b/tests/streaming/test_streaming.py
index e3283765..e9befaa3 100644
--- a/tests/streaming/test_streaming.py
+++ b/tests/streaming/test_streaming.py
@@ -1,4 +1,6 @@
+import asyncio
 import os
+import time
 from unittest import mock
 from unittest.mock import AsyncMock
 
@@ -649,3 +651,59 @@ async def test_stream_listener_returns_correct_chunk_json_adapter_untokenized_st
         assert all_chunks[1].predict_name == "predict2"
         assert all_chunks[1].signature_field_name == "judgement"
         assert all_chunks[1].chunk == "The answer provides a humorous and relevant punchline to the classic joke setup."
+
+
+@pytest.mark.anyio
+async def test_status_message_non_blocking():
+    def dummy_tool():
+        time.sleep(1)
+        return "dummy_tool_output"
+
+    class MyProgram(dspy.Module):
+        def forward(self, question, **kwargs):
+            dspy.Tool(dummy_tool)()
+            return dspy.Prediction(answer="dummy_tool_output")
+
+    program = dspy.streamify(MyProgram(), status_message_provider=StatusMessageProvider())
+
+    with mock.patch("litellm.acompletion", new_callable=AsyncMock, side_effect=[dummy_tool]):
+        with dspy.context(lm=dspy.LM("openai/gpt-4o-mini", cache=False)):
+            output = program(question="why did a chicken cross the kitchen?")
+            timestamps = []
+            async for value in output:
+                if isinstance(value, dspy.streaming.StatusMessage):
+                    timestamps.append(time.time())
+
+    # timestamps[0]: tool start message
+    # timestamps[1]: tool end message
+    # There should be ~1 second delay between the tool start and end messages because we explicitly sleep for 1 second
+    # in the tool.
+    assert timestamps[1] - timestamps[0] >= 1
+
+
+@pytest.mark.anyio
+async def test_status_message_non_blocking_async_program():
+    async def dummy_tool():
+        await asyncio.sleep(1)
+        return "dummy_tool_output"
+
+    class MyProgram(dspy.Module):
+        async def aforward(self, question, **kwargs):
+            await dspy.Tool(dummy_tool).acall()
+            return dspy.Prediction(answer="dummy_tool_output")
+
+    program = dspy.streamify(MyProgram(), status_message_provider=StatusMessageProvider(), is_async_program=True)
+
+    with mock.patch("litellm.acompletion", new_callable=AsyncMock, side_effect=[dummy_tool]):
+        with dspy.context(lm=dspy.LM("openai/gpt-4o-mini", cache=False)):
+            output = program(question="why did a chicken cross the kitchen?")
+            timestamps = []
+            async for value in output:
+                if isinstance(value, dspy.streaming.StatusMessage):
+                    timestamps.append(time.time())
+
+    # timestamps[0]: tool start message
+    # timestamps[1]: tool end message
+    # There should be ~1 second delay between the tool start and end messages because we explicitly sleep for 1 second
+    # in the tool.
+    assert timestamps[1] - timestamps[0] >= 1