Repository: stanfordnlp/dspy (python)

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

Pull request:
Fix the error that status message streaming was blocking 
resolve #8424 

The old status communication was somehow blocking the event loop, which resulted in the status start/end message to come at the same time. 

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