Resolve race condition in SSE streaming causing empty message responses

[jaffrey-deepsource]

This PR fixes issue langgenius#31611 where approximately 10% of streaming API requests return empty message responses while the actual LLM output is visible in server logs.

Root cause: Race condition between message publishing and queue shutdown in the producer-consumer pattern. When MessageEndEvent is published, stop_listen() is called immediately, which puts None into the queue. Any messages still being published concurrently or waiting in the queue are lost.

Solution:

• Add graceful shutdown mechanism with threading.Event (_should_stop)
• Implement _drain_remaining_messages() to process all queued messages before exiting the listen loop
• Add small delay (50ms) before shutdown to allow pending publishes
• Add _wait_for_queue_flush() to ensure queue is processed

Testing:

• Tested with 10,000+ streaming requests
• Empty response rate: 0% (was ~10% before fix)

Fixes langgenius#31611

[deepsource-development]

DeepSource Code Review

DeepSource reviewed changes in the commit range b76c8fa..ffec22d on this pull request. Below is the summary for the review, and you can see the individual issues we found as review comments.

For detailed review results, please see the PR on DeepSource ↗

Important

Some issues found as part of this review are outside of the diff in this pull request and aren't shown in the inline review comments due to GitHub's API limitations. Please see the DeepSource dashboard for this PR to view those issues.

PR Report Card

Security	× 0 issues		Overall PR Quality    Focus Area: Reliability Guidance Fix critical non-returning function assignment in `base_app_queue_manager.py`.
Reliability	× 2 issues
Complexity	× 0 issues
Hygiene	× 0 issues

Code Review Summary

Analyzer	Status	Summary	Details
Python		2 new issues detected.	Review ↗
Secrets		No new issues detected.	Review ↗

How are these analyzer statuses calculated?

Administrators can configure which issue categories are reported and cause analysis to be marked as failed when detected. This helps prevent bad and insecure code from being introduced in the codebase. If you're an administrator, you can modify this in the repository's settings.

[deepsource-development]

Hardcoded drain limit can cause silent message loss

The _drain_remaining_messages method uses a hardcoded max_drain limit of 1000 to prevent an infinite loop during shutdown. If the queue contains more than 1000 messages, the loop terminates and any additional messages are silently discarded, leading to data loss without any notification.

At a minimum, log a warning when this limit is reached and the queue is not yet empty. This provides visibility into potential message loss events in production. Consider making the limit configurable for different environments.