Fix dropped first chunk in incremental stream readers#339
Open
MaxHeimbrock wants to merge 1 commit into
Open
Conversation
ReadIncremental() sent the FFI request before constructing the ReadIncrementalInstruction that subscribes to reader events. Chunk events are delivered directly on the FFI callback thread, and the FFI server emits already-buffered chunks as soon as it receives the request, so any chunk emitted before the subscription existed was silently lost. For delta text streams (e.g. agent transcriptions) this intermittently dropped the first word(s) of a reply. Subscribe first, then send — matching the order ReadAll() already uses. Applies to both TextStreamReader and ByteStreamReader. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
Agent transcriptions were intermittently missing the first word(s) of a reply.
TextStreamReader.ReadIncremental()(and the byte-stream equivalent) sent the FFI request before constructing theReadIncrementalInstructionwhose constructor subscribes toTextStreamReaderEventReceived. Three facts turn that ordering into a real chunk drop:read_incrementalrequest (livekit-ffi/src/server/data_stream.rs). By the time the app callsReadIncremental(), the first chunk of an agent reply has typically already arrived and is buffered — the stream-open event took a main-thread hop to reach the handler.FFIClient.RouteFfiEventinvokesTextStreamReaderEventReceiveddirectly on the FFI callback thread (no main-thread queueing), so the first chunk event races the remainder ofReadIncremental()on the main thread.ReadIncrementalInstructionBaseonly buffers chunks that arrive after subscription.When the Rust task won the race, the first chunk was gone. For delta text streams (agent transcriptions via
lk.transcription) that means the first word(s) of the reply never render.ReadAll()already used the safe order, which is why snapshot-style streams were unaffected.Fix
Construct the instruction (subscribing to reader events) before sending the request, in both
TextStreamReader.ReadIncremental()andByteStreamReader.ReadIncremental(). C# event add/invoke is an atomic delegate swap, so subscribing first is sufficient; chunks arriving before the first yield are already handled by the pending-chunk queue.🤖 Generated with Claude Code