with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(run_in_new_loop)
    return future.result()  # ← BLOCKS the main event loop!

return syncify(_send)()  # ← Raises AnyIO error without proper context

import asyncio
import dspy
from dspy.streaming import StreamListener

# 1. Set up streaming with async tool
lm = dspy.LM(model="openai/gpt-4o")

def my_async_tool(query: str) -> str:
    """An async tool that simulates work."""
    import time
    time.sleep(0.1)  # Simulate async work
    return f"Result for: {query}"

signature = dspy.Signature(
    {"input": (str, dspy.InputField()), "output": (str, dspy.OutputField())}
)

# 2. Create ReAct program with tools
program = dspy.ReAct(signature, tools=[my_async_tool], max_iters=2)

# 3. Wrap with streamify
streaming_task = dspy.streamify(
    program,
    is_async_program=True,
    stream_listeners=[StreamListener(signature_field_name="output")]
)

# 4. Execute - this will freeze after tool execution
async def run():
    with dspy.context(lm=lm):
        stream = streaming_task(input="test query")
        async for value in stream:
            print(value)

asyncio.run(run())
# ⚠️ Stream freezes at "Tool calling finished!" - never completes

def sync_send_to_stream(stream, message):
    """Send message to stream without blocking the event loop."""

    async def _send():
        try:
            await stream.send(message)
        except Exception:
            pass  # Gracefully handle send failures

    try:
        loop = asyncio.get_running_loop()
        # Non-blocking: schedule as background task (fire-and-forget)
        loop.create_task(_send())
    except RuntimeError:
        # No event loop running - safe to create new one
        try:
            asyncio.run(_send())
        except Exception:
            pass  # Gracefully handle failures