fix(dspy): Reset stream listener state per streamify call for inter-call reuse

[detail-app]

Warning

GitHub issue creation failed

Detail attempted to publish this bug to GitHub, but the issue could not be created. This fix PR was created without that issue, and missing tracker references are shown as Unknown issue.

You can review and merge this PR normally. Please review your tracker integration settings before the next publish run.

Detail bug report: View on Detail

Closes Unknown issue

📝 Changes Description

This MR/PR contains the following changes:

• Bug: dspy.streamify captures its StreamListener objects once, outside async_streamer. At the end of the first call, listener.finalize() sets listener.stream_end = True and never resets it. On any subsequent call, StreamListener.receive() early-returns on every chunk (stream_end is True, allow_reuse defaults to False), silently dropping all incremental StreamResponse chunks (the final Prediction is still delivered, so the loss is invisible). With include_final_prediction_in_output_stream=False at construction, the second call yields an empty stream (no chunks, no Prediction), because the "yield final Prediction?" guard depends on stale listener state.

• Fix: Add StreamListener.reset() to clear per-stream run-time state (stream_start/stream_end/cache_hit flags, field_start_queue, field_end_queue, json_adapter_state["field_accumulated_messages"]) while preserving configuration (signature_field_name, predict, predict_name, allow_reuse, adapter_identifiers). async_streamer now calls listener.reset() for every listener at the start of each invocation, so a streamify-returned streamer is reusable by default — the natural usage for long-lived callers (web handlers, REPLs, notebook loops). The allow_reuse=True branch of receive() is refactored to call self.reset() (behavior-preserving for the documented intra-run reuse, e.g. dspy.ReAct loops); the sync path (async_streaming=False) inherits the fix via async_streamer. This was chosen over a "fail loudly / document as single-use" alternative because the streamer is a plain callable that is natural to reuse, and the docs never state it is single-use.

✅ Contributor Checklist

• Pre-Commit checks are passing (locally) — ruff check (the hook's action) passes; ruff format --check clean on all three files.
• Title of your PR / MR corresponds to the required format
• Commit message follows required format {label}(dspy): {message}

⚠️ Warnings

• AI-assisted disclosure: Authored by Detail (automatic bug-fix agent). The prompt was the full streaming inter-call reuse bug report. All changes were verified by the operator (every test/procedure below was run).
• No functional behavior changes for single-call usage; the first call of any streamer behaves exactly as before.

---

Automatic Fixes PRs can be configured here.

[greptile-apps]

Greptile Summary

This PR resets listener parsing state at the beginning of each streamify invocation so sequential calls can continue emitting incremental responses.

• Adds a centralized StreamListener.reset() operation.
• Applies the reset to asynchronous and, transitively, synchronous streamer invocations.
• Adds regression coverage for repeated calls, output suppression, content consistency, and reusable listeners.
• The captured listeners remain shared across overlapping invocations, allowing one request to reset another request’s active parsing state.

Confidence Score: 4/5

This PR is not yet safe to merge because overlapping uses of a shared streamer can reset and corrupt one another’s active listener state.

Sequential reuse is repaired and the reset itself is complete, but listener instances remain shared by all invocations; the new unconditional reset can erase an in-progress invocation’s parser state when another request starts.

Files Needing Attention: dspy/streaming/streamify.py, tests/streaming/test_streaming.py

Important Files Changed

Filename	Overview
dspy/streaming/streamify.py	Resets captured listeners per invocation, fixing sequential reuse but introducing destructive interference between overlapping calls.
dspy/streaming/streaming_listener.py	Centralizes complete runtime-state cleanup in reset() while preserving listener configuration.
tests/streaming/test_streaming.py	Adds strong sequential-reuse coverage, but lacks overlapping-call coverage and contains one inaccurate test comment.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    S[streamify construction] --> L[Captured shared StreamListener]
    S --> A[Invocation A]
    S --> B[Invocation B]
    A --> RA[Reset listener]
    RA --> PA[Parse A chunks into listener state]
    B --> RB[Reset same listener]
    RB --> C[Clear A flags, queues, and accumulated JSON]
    C --> X[Missing, duplicated, or mixed streamed output]

Loading

_{Reviews (1): Last reviewed commit: "fix(dspy): Reset stream listener state p..." | Re-trigger Greptile}

[greptile-apps]

Shared Reset Corrupts Active Streams

When the same reusable streamer handles overlapping calls, both calls use the captured StreamListener objects. Starting the second call resets the first call's active parsing flags, queues, and accumulated JSON. A module-level streamer serving concurrent requests can therefore drop, duplicate, or mix chunks between streams. Listener state needs to be isolated per invocation rather than reset in place.

[greptile-apps]

Comment Contradicts Expected Output

This comment says the suppressed-output configuration streams the final Prediction, but the test correctly expects only seven incremental values. This documents the opposite of the tested contract and could mislead future maintainers.

Suggested change

	# state. After the fix, every call should stream the field chunks and the final Prediction.
	# state. After the fix, every call should stream the field chunks without the final Prediction.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!