[Bug] Abandoning a partially consumed sync streamify stream silently consumes the whole upstream in a background thread

What happened?

dspy.streamify(program, async_streaming=False) returns apply_sync_streaming(async_streamer(...)) (dspy/streaming/streamify.py:238). The sync generator is fed by a background daemon thread that pumps the async stream into an unbounded Queue:

def producer():
    async def runner():
        try:
            async for item in async_generator:
                queue.put(item)
        except BaseException as exc:
            queue.put((exception_sentinel, exc))
        finally:
            queue.put(stop_sentinel)

    context.run(asyncio.run, runner())

thread = threading.Thread(target=producer, daemon=True)

Nothing ever tells that thread to stop. If the consumer walks away from a partially consumed stream — break out of the loop, gen.close(), or just dropping the reference — the producer keeps iterating async_streamer to the very end. The wrapped program (and its LM call) runs to completion in the background:

• the entire completion is still consumed — every remaining token is pulled off the provider stream and billed, for output nobody reads;
• every chunk is buffered into the unbounded queue, held until process exit (the daemon thread and its queue leak per abandoned stream);
• async_streamer.aclose() is never called on the early-exit path, so the anyio task group is torn down by exhaustion, never by cancellation.

This is the sync twin of #10380 (which is about aclose() on the async generator raising): the async side crashed loudly; the sync side fails silently, which is arguably worse — nothing in the program's observable behavior says the stream is still running.

To reproduce

Offline, no network (the counting generator stands in for the LM stream):

import time
from dspy.streaming.streamify import apply_sync_streaming

consumed = {"count": 0, "finished": False}

async def upstream():
    for i in range(1000):
        consumed["count"] = i + 1
        yield f"chunk-{i}"
    consumed["finished"] = True

gen = apply_sync_streaming(upstream())
next(gen)          # consume ONE chunk
gen.close()        # walk away
time.sleep(0.5)
print(consumed)    # {'count': 1000, 'finished': True}

Observed on current main (4368715b): after consuming one chunk and closing, the upstream has been consumed 1000/1000 — the producer had already raced through the whole stream by the time close() returned, and threading.enumerate() still lists the producer thread. With a slow (network-paced) upstream the effect is the same, just spread over the stream's full duration: close() returns immediately while the background thread keeps pulling chunks until the provider finishes.

Expected behavior

Closing (or abandoning) a partially consumed sync stream stops the background consumption: the producer task is cancelled, the async generator is closed, the thread exits, and no further upstream chunks are pulled. Full consumption and error propagation (#9142) behave exactly as today.

Environment

• DSPy main (4368715b), also reproduces on 3.4.0b1
• Python 3.12, macOS/Linux

For Agents

Implementation sketch, one PR:

• In apply_sync_streaming, expose the runner's loop/task to the consumer thread (set from inside runner() before iterating; an Event to signal readiness).
• Wrap the consumer loop in try/finally. When the finally runs before the stop sentinel was seen (early close/GC), set a close_requested flag, loop.call_soon_threadsafe(task.cancel), and thread.join with a bounded timeout.
• In the producer: a CancelledError after close_requested is the requested shutdown, not an error to report; anything else keeps propagating through the exception sentinel (#9142 stays fixed). In its finally, await async_generator.aclose() (tolerating the task-group re-raise from #10380 until that fix lands) and always put the stop sentinel so a concurrent reader cannot block forever.
• Acceptance:
  • closing after a partial read stops upstream consumption (a counting fake advances by at most a chunk or two after close() returns) and the producer thread terminates;
  • a fully consumed stream and close() after exhaustion behave exactly as today;
  • producer exceptions still surface to the consumer (existing behavior);
  • no change to the async path.