Fix for incremental reader chunk-drop

Runtime/Scripts/DataStream.cs

@@ -77,6 +77,7 @@ internal StreamError(Proto.StreamError proto) : base(proto.Description) { }
     public abstract class ReadIncrementalInstructionBase<TContent> : StreamYieldInstruction
     {
         private readonly ulong _handleValue;
+        private readonly Queue<TContent> _pendingChunks = new();
         private TContent _latestChunk;
 
         /// <summary>
@@ -107,8 +108,26 @@ protected ReadIncrementalInstructionBase(FfiHandle readerHandle)
 
         protected void OnChunk(TContent content)
         {
-            _latestChunk = content;
-            IsCurrentReadDone = true;
+            if (IsCurrentReadDone)
+            {
+                // Consumer hasn't yielded since the last chunk; buffer until Reset().
+                _pendingChunks.Enqueue(content);
+            }
+            else
+            {
+                _latestChunk = content;
+                IsCurrentReadDone = true;
+            }
+        }
+
+        public override void Reset()
+        {
+            base.Reset();
+            if (_pendingChunks.Count > 0)
+            {
+                _latestChunk = _pendingChunks.Dequeue();
+                IsCurrentReadDone = true;
+            }
         }
 
         protected void OnEos(Proto.StreamError protoError)

Tests/EditMode/DataStreamIncrementalReadTests.cs

@@ -0,0 +1,44 @@
+using System;
+using System.Collections.Generic;
+using LiveKit.Internal;
+using NUnit.Framework;
+
+namespace LiveKit.EditModeTests
+{
+    public class DataStreamTests
+    {
+        private sealed class TestIncrementalReader : ReadIncrementalInstructionBase<string>
+        {
+            public TestIncrementalReader(FfiHandle h) : base(h) { }
+            public void PushChunk(string content) => OnChunk(content);
+            public string Value => LatestChunk;
+        }
+
+        [Test]
+        public void OnChunk_MultipleChunksBeforeConsumerDrains_AllChunksAreObserved()
+        {
+            using var handle = new FfiHandle(IntPtr.Zero);
+            var reader = new TestIncrementalReader(handle);
+
+            // Three ChunkReceived events arrive back-to-back on the sync context,
+            // before the consumer coroutine resumes from its yield.
+            reader.PushChunk("A");
+            reader.PushChunk("B");
+            reader.PushChunk("C");
+
+            // Drain like a consumer would: read the current chunk, Reset, loop
+            // while more is readable.
+            var observed = new List<string>();
+            while (!reader.keepWaiting)
+            {
+                observed.Add(reader.Value);
+                if (reader.IsEos) break;
+                reader.Reset();
+            }
+
+            CollectionAssert.AreEqual(new[] { "A", "B", "C" }, observed,
+                "ReadIncrementalInstructionBase dropped chunks when multiple " +
+                "ChunkReceived events arrived before the consumer could yield.");
+        }
+    }
+}