test(worker): deflake EventBurst_DrainLoopCoalescesFlushes

Root cause. The test sampled its flush baseline after a bare `Task.Delay(100)`
and then charged every later flush to the 128-event burst. Two facts make that
window unsound:

1. `RunHeartbeatLoopAsync` sends its first beat *immediately* on entering the
   message loop — the far-off `HeartbeatInterval` only spaces later beats — so
   the pre-burst frames are WorkerHello, WorkerReady, and a heartbeat, not the
   two the test's comment assumed.
2. `WorkerFrameWriter.DrainQueuedFramesAsync` flushes *after* writing a drained
   batch, so the bytes reach the pipe before the flush runs. Reading a frame off
   the gateway side is therefore no evidence that its flush has been counted,
   and no sleep makes it evidence.

Under load on the shared windows-x86 runner the first heartbeat's flush was
scheduled after the 100 ms sample, so it landed inside the measured window and
the assertion saw two flushes for the burst — exactly the observed
`Expected: 1 / Actual: 2` (Gitea run 675, job 2558, and the same failure since
a346d51). Nothing about the coalescing behavior was wrong; only the test's
timing assumption.

Fix. Replace the sleep with explicit synchronization, no widened timeouts.
`FlushCountingPassthroughStream` now records the flush *shape* — the number of
stream writes coalesced into each flush — and exposes
`WaitForAllWritesFlushedAsync`, a TaskCompletionSource signal released when the
next flush drains the pending writes. The test reads the first heartbeat (the
last pre-burst frame), waits for its flush, and only then samples the baseline;
after the burst it takes the same edge before asserting. The assertion is also
sharpened from a bare count to the shape: exactly one flush beyond the baseline
*and* that flush carried all 128 event frames, so a split batch fails even if
the reader observes it mid-split.

docs/WorkerFrameProtocol.md notes the peer-visible ordering the fix turns on:
frames reach the pipe before the flush that follows them, so an observer of the
flush must wait for it rather than infer it from frames arriving.

docs/WorkerFrameProtocol.md

@@ -123,7 +123,11 @@ runs after the whole batch, and only then does every successfully-written
 frame's completion resolve — so a caller's `WriteAsync` still does not
 complete until its bytes are both written *and* flushed, but a batch that
 happened to contain several queued frames pays one flush instead of one per
-frame. The event drain loop (`WorkerPipeSession.RunEventDrainLoopAsync`)
+frame. Note the ordering this implies at the peer: the frames reach the pipe
+before the flush that follows them, so the gateway can read a whole batch
+while the writer has not yet flushed it. Anything observing the flush itself
+(a test counting flushes, for instance) must wait for the flush, not infer it
+from frames arriving. The event drain loop (`WorkerPipeSession.RunEventDrainLoopAsync`)
 submits a whole drained event batch through `WriteBatchAsync`, which enqueues
 every frame under one `_gate` acquisition, takes the write lock once, and
 drains them together, so a burst of N events costs one flush rather than N —

src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs

@@ -1214,7 +1214,9 @@ public sealed class WorkerPipeSessionTests
         using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15));
         using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
         FakeRuntimeSession runtime = new();
-        // A far-off heartbeat interval keeps heartbeat flushes out of the measurement window.
+        // A far-off heartbeat interval keeps every heartbeat after the first out of the measurement
+        // window; the first beat is sent immediately on entering the message loop (see
+        // RunHeartbeatLoopAsync) and is closed out explicitly below.
         FlushCountingPassthroughStream countingStream = new(pipePair.WorkerStream);
         WorkerPipeSession session = CreatePipeSession(
             countingStream,
@@ -1227,8 +1229,18 @@ public sealed class WorkerPipeSessionTests
         Task runTask = session.RunAsync(cancellation.Token);
         await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
 
-        // Let the idle drain loop settle (no events yet → no flushes) and record the baseline.
-        await Task.Delay(100, cancellation.Token);
+        // Close the pre-burst window on an explicit signal rather than a sleep. Three frames precede
+        // the burst — WorkerHello, WorkerReady, and the immediate first heartbeat — and each is
+        // flushed only after its bytes are already on the pipe, so having read a frame is no evidence
+        // that its flush has been counted. Read the first beat (the last pre-burst frame), then wait
+        // for the writer's deferred flush before sampling the baseline. A fixed delay left the first
+        // beat's flush free to land after the sample under CI scheduling pressure, where it was
+        // charged to the burst and the assertion below saw two flushes instead of one.
+        await ReadUntilAsync(
+            pipePair.GatewayReader,
+            WorkerEnvelope.BodyOneofCase.WorkerHeartbeat,
+            cancellation.Token);
+        await countingStream.WaitForAllWritesFlushedAsync(cancellation.Token);
         int baselineFlushes = countingStream.FlushCount;
 
         // Enqueue a full 128-event batch atomically so the drain loop sees it as one batch.
@@ -1250,8 +1262,15 @@ public sealed class WorkerPipeSessionTests
                 cancellation.Token);
         }
 
-        // The whole burst cost exactly one additional flush.
-        Assert.Equal(1, countingStream.FlushCount - baselineFlushes);
+        // Take the same edge on the burst's own flush — the 128 frames are on the wire before the
+        // writer flushes them — then assert on the recorded flush shape: exactly one flush beyond the
+        // baseline, and that flush carried every event frame of the burst. Asserting the shape (not
+        // just the count) is what makes the coalescing claim faithful: a split batch would show its
+        // first flush carrying fewer than the whole burst.
+        await countingStream.WaitForAllWritesFlushedAsync(cancellation.Token);
+        IReadOnlyList<int> flushWriteCounts = countingStream.SnapshotFlushWriteCounts();
+        Assert.Equal(baselineFlushes + 1, flushWriteCounts.Count);
+        Assert.Equal(burst, flushWriteCounts[baselineFlushes]);
 
         await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
     }
@@ -2154,13 +2173,22 @@ public sealed class WorkerPipeSessionTests
         }
     }
 
-    // Wraps the worker side of the pipe and counts FlushAsync calls so a test can assert the event
-    // drain loop coalesces a burst into a single flush. Delegates every other operation to the inner
-    // stream; does not own the inner stream's lifetime (PipePair disposes it).
+    // Wraps the worker side of the pipe and records the flush shape of the frames the writer emits —
+    // how many stream writes each flush coalesced — so a test can assert the event drain loop turns a
+    // burst into a single flush. Delegates every other operation to the inner stream; does not own the
+    // inner stream's lifetime (PipePair disposes it).
+    //
+    // The writer flushes only after writing a whole drained batch, so the peer can read every frame of
+    // that batch before the flush runs. Neither "I read the frames" nor any fixed sleep is evidence
+    // that a flush has been counted; WaitForAllWritesFlushedAsync is the explicit edge a test must
+    // take before sampling the counters.
     private sealed class FlushCountingPassthroughStream : Stream
     {
         private readonly Stream inner;
-        private int flushCount;
+        private readonly object gate = new();
+        private readonly List<int> flushWriteCounts = new();
+        private readonly List<FlushWaiter> waiters = new();
+        private int writesSinceLastFlush;
 
         /// <summary>Initializes the passthrough over the given inner stream.</summary>
         /// <param name="inner">The stream to delegate to.</param>
@@ -2169,8 +2197,53 @@ public sealed class WorkerPipeSessionTests
             this.inner = inner;
         }
 
-        /// <summary>Gets the number of <see cref="FlushAsync"/> calls observed so far.</summary>
-        public int FlushCount => Volatile.Read(ref flushCount);
+        /// <summary>Gets the number of flushes observed so far.</summary>
+        public int FlushCount
+        {
+            get
+            {
+                lock (gate)
+                {
+                    return flushWriteCounts.Count;
+                }
+            }
+        }
+
+        /// <summary>
+        ///     Returns the number of stream writes coalesced into each flush, in flush order.
+        /// </summary>
+        /// <returns>A snapshot of the per-flush write counts.</returns>
+        public IReadOnlyList<int> SnapshotFlushWriteCounts()
+        {
+            lock (gate)
+            {
+                return flushWriteCounts.ToArray();
+            }
+        }
+
+        /// <summary>
+        ///     Completes once every write issued so far has been flushed, giving the caller a
+        ///     happens-before edge on the writer's deferred flush instead of a timing guess.
+        /// </summary>
+        /// <param name="cancellationToken">Token to abandon the wait.</param>
+        /// <returns>A task that completes when no write is left unflushed.</returns>
+        public Task WaitForAllWritesFlushedAsync(CancellationToken cancellationToken)
+        {
+            FlushWaiter waiter;
+            lock (gate)
+            {
+                if (writesSinceLastFlush == 0)
+                {
+                    return Task.CompletedTask;
+                }
+
+                // Any flush drains every pending write, so the next flush is exactly the edge wanted.
+                waiter = new FlushWaiter(flushWriteCounts.Count + 1);
+                waiters.Add(waiter);
+            }
+
+            return AwaitFlushWaiterAsync(waiter, cancellationToken);
+        }
 
         /// <inheritdoc />
         public override bool CanRead => inner.CanRead;
@@ -2194,14 +2267,14 @@ public sealed class WorkerPipeSessionTests
         /// <inheritdoc />
         public override void Flush()
         {
-            Interlocked.Increment(ref flushCount);
+            RecordFlush();
             inner.Flush();
         }
 
         /// <inheritdoc />
         public override Task FlushAsync(CancellationToken cancellationToken)
         {
-            Interlocked.Increment(ref flushCount);
+            RecordFlush();
             return inner.FlushAsync(cancellationToken);
         }
 
@@ -2219,11 +2292,87 @@ public sealed class WorkerPipeSessionTests
         public override void SetLength(long value) => inner.SetLength(value);
 
         /// <inheritdoc />
-        public override void Write(byte[] buffer, int offset, int count) => inner.Write(buffer, offset, count);
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            RecordWrite();
+            inner.Write(buffer, offset, count);
+        }
 
         /// <inheritdoc />
         public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
-            => inner.WriteAsync(buffer, offset, count, cancellationToken);
+        {
+            RecordWrite();
+            return inner.WriteAsync(buffer, offset, count, cancellationToken);
+        }
+
+        private static async Task AwaitFlushWaiterAsync(
+            FlushWaiter waiter,
+            CancellationToken cancellationToken)
+        {
+            using (cancellationToken.Register(() => waiter.Completion.TrySetCanceled()))
+            {
+                await waiter.Completion.Task.ConfigureAwait(false);
+            }
+        }
+
+        // Counted at write issue, not completion: the peer can observe the bytes as soon as the write
+        // is issued, so the pending count must already reflect the write by then.
+        private void RecordWrite()
+        {
+            lock (gate)
+            {
+                writesSinceLastFlush++;
+            }
+        }
+
+        private void RecordFlush()
+        {
+            List<FlushWaiter>? released = null;
+            lock (gate)
+            {
+                flushWriteCounts.Add(writesSinceLastFlush);
+                writesSinceLastFlush = 0;
+                for (int index = waiters.Count - 1; index >= 0; index--)
+                {
+                    if (waiters[index].TargetFlushCount <= flushWriteCounts.Count)
+                    {
+                        released ??= new List<FlushWaiter>();
+                        released.Add(waiters[index]);
+                        waiters.RemoveAt(index);
+                    }
+                }
+            }
+
+            if (released is null)
+            {
+                return;
+            }
+
+            foreach (FlushWaiter waiter in released)
+            {
+                waiter.Completion.TrySetResult(true);
+            }
+        }
+
+        // A pending WaitForAllWritesFlushedAsync call: completes once the recorded flush count
+        // reaches TargetFlushCount.
+        private sealed class FlushWaiter
+        {
+            /// <summary>Initializes a waiter released at the given flush ordinal.</summary>
+            /// <param name="targetFlushCount">Flush count that releases the waiter.</param>
+            public FlushWaiter(int targetFlushCount)
+            {
+                TargetFlushCount = targetFlushCount;
+                Completion = new TaskCompletionSource<bool>(
+                    TaskCreationOptions.RunContinuationsAsynchronously);
+            }
+
+            /// <summary>Gets the flush count at which this waiter completes.</summary>
+            public int TargetFlushCount { get; }
+
+            /// <summary>Gets the completion signaled when the target flush count is reached.</summary>
+            public TaskCompletionSource<bool> Completion { get; }
+        }
     }
 
     private sealed class PipePair : IDisposable