fix(WRK-21): make drain budget monotonic at the reserve boundary; guard the reply-too-large fallback

Review follow-ups on the WRK-21 cluster.

1. ResolveDrainReplyByteBudget was a step function, not a floor: just above the
   64 KiB reserve the budget collapsed to a few bytes (at the validator-permitted
   floor MaxMessageBytes = 1024 + 64 KiB it was exactly 1024), too small to move a
   byte-heavy event, so DrainEvents truncated on every call and the drain-until-
   empty loop never terminated. It now takes the max of (frameMax - reserve) and
   frameMax/2, so the budget is monotonic and never below half the frame max. New
   test DrainEvents_AtValidatorFloorFrameMax_MakesProgressAndTerminates drives a
   byte-heavy queue at the exact validator floor and asserts it drains to empty
   with no head ever reported oversized.

2. The reply-too-large fallback write is now itself size-guarded
   (WriteReplyTooLargeFallbackAsync, used by both the control and STA reply seams):
   at a pathologically tiny negotiated max below the gateway's floor the fallback
   could also throw MessageTooLarge and — uncaught — kill the session, defeating the
   "no diagnostics command is session-fatal" invariant. It now log-and-swallows;
   comment notes WRK-24 adds the negotiated-max lower bound that makes it unreachable.

3. Corrected the RepeatedFieldOverheadBytes doc comments: WorkerEvent.CalculateSize()
   already includes the event's tag and length prefix (the same shape the reply's
   repeated events field packs), so the 8 bytes is pure slack over an already-
   conservative estimate, not compensation for a missing wrapper.

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

@@ -633,6 +633,76 @@ public sealed class WorkerPipeSessionTests
         await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
     }
 
+    /// <summary>
+    ///     Regression for the reserve-boundary budget bug. The gateway validator accepts a Worker
+    ///     frame maximum as low as 1024 + 64 KiB, and just above that boundary a naive
+    ///     subtract-then-guard budget collapses to ~1024 bytes — too small to move even one
+    ///     byte-heavy event, so every drain reports truncation with the same head blocked and the
+    ///     drain-until-empty loop never terminates. The budget is now a floor (never below half the
+    ///     negotiated maximum), so a byte-heavy queue drains to empty even at the validator floor.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DrainEvents_AtValidatorFloorFrameMax_MakesProgressAndTerminates()
+    {
+        // The lowest Worker.MaxMessageBytes GatewayOptionsValidator permits: the public gRPC floor
+        // (1024) plus the 64 KiB envelope-overhead reserve. The naive budget would be exactly 1024
+        // here; the floored budget is half of the frame max (~33 KiB).
+        const uint validatorFloorFrameMax = 1024 + (64 * 1024);
+        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(60));
+        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+        FakeRuntimeSession runtime = new()
+        {
+            SuppressDrainForBatchSize = 128,
+            BackingQueue = CreateByteHeavyQueue(200, ByteHeavyEventPayloadBytes),
+        };
+        WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime);
+        Task runTask = session.RunAsync(cancellation.Token);
+        await CompleteGatewayHandshakeAsync(pipePair, validatorFloorFrameMax, cancellation.Token);
+
+        int recovered = 0;
+        int replyCount = 0;
+        while (true)
+        {
+            string correlationId = $"floor-drain-{replyCount}";
+            await pipePair.GatewayWriter
+                .WriteAsync(
+                    CreateControlCommandEnvelope(
+                        correlationId,
+                        MxCommandKind.DrainEvents,
+                        command => command.DrainEvents = new DrainEventsCommand { MaxEvents = 0 }),
+                    cancellation.Token);
+
+            WorkerEnvelope replyEnvelope = await ReadUntilAsync(
+                pipePair.GatewayReader,
+                WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
+                envelope => envelope.WorkerCommandReply.Reply.CorrelationId == correlationId,
+                cancellation.Token);
+            replyCount++;
+
+            MxCommandReply reply = replyEnvelope.WorkerCommandReply.Reply;
+            Assert.True(
+                replyEnvelope.CalculateSize() <= validatorFloorFrameMax,
+                $"reply {replyCount} serialized to {replyEnvelope.CalculateSize()} bytes.");
+
+            int drainedThisReply = reply.DrainEvents.Events.Count;
+            if (drainedThisReply == 0)
+            {
+                break;
+            }
+
+            // The head is never reported as oversized at this frame max: the ~33 KiB floored budget
+            // comfortably fits the ~1.8 KiB events, so each reply makes real progress.
+            Assert.DoesNotContain("alone exceeds", reply.DiagnosticMessage);
+            recovered += drainedThisReply;
+            Assert.True(replyCount < 200, "DrainEvents made no progress at the validator floor frame max.");
+        }
+
+        Assert.Equal(200, recovered);
+
+        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
+    }
+
     /// <summary>
     ///     Verifies the control-reply write seam is not session-fatal on size (WRK-21 backstop). The
     ///     reply builders size their payloads, so this path needs a deliberately budget-blind drain

src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs

@@ -587,10 +587,45 @@ public sealed class WorkerPipeSession
             when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge)
         {
             LogControlReplyTooLarge(correlationId, kind, exception);
+            await WriteReplyTooLargeFallbackAsync(correlationId, kind, cancellationToken)
+                .ConfigureAwait(false);
+        }
+    }
+
+    /// <summary>
+    ///     Writes the small <c>InvalidRequest</c> reply that answers a correlation whose real reply
+    ///     overshot the frame maximum. The fallback itself is a handful of bytes, so it fits any
+    ///     sane negotiated maximum; the only way it can also throw <c>MessageTooLarge</c> is a
+    ///     pathologically tiny negotiated maximum below the gateway's validation floor — the
+    ///     pre-existing WRK-24 gap, which adds the negotiated-max lower bound that makes this
+    ///     unreachable. Until then, a defensive swallow keeps the "no diagnostics command is
+    ///     session-fatal" invariant true even in that degenerate config: the correlation goes
+    ///     unanswered and the gateway's own per-command timeout covers it, but the session lives.
+    /// </summary>
+    private async Task WriteReplyTooLargeFallbackAsync(
+        string correlationId,
+        MxCommandKind kind,
+        CancellationToken cancellationToken)
+    {
+        try
+        {
             await WriteControlReplyAsync(
                 CreateReplyTooLargeReply(correlationId, kind),
                 cancellationToken).ConfigureAwait(false);
         }
+        catch (WorkerFrameProtocolException exception)
+            when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge)
+        {
+            _logger?.Error(
+                "WorkerControlReplyFallbackTooLarge",
+                new Dictionary<string, object?>
+                {
+                    ["correlation_id"] = correlationId,
+                    ["command_kind"] = kind.ToString(),
+                    ["max_message_bytes"] = _options.MaxMessageBytes,
+                    ["reason"] = exception.Message,
+                });
+        }
     }
 
     private void LogControlReplyTooLarge(
@@ -715,15 +750,21 @@ public sealed class WorkerPipeSession
 
     /// <summary>
     ///     Byte budget for the events packed into one DrainEvents reply: the negotiated frame
-    ///     maximum less a fixed reserve for the envelope/reply wrapper. A negotiated maximum below
-    ///     the reserve would otherwise yield a non-positive budget and stall the drain forever, so
-    ///     a tiny frame maximum falls back to half of itself — still ample headroom for a wrapper
-    ///     measured in tens of bytes.
+    ///     maximum less a fixed reserve for the envelope/reply wrapper, but never below half the
+    ///     negotiated maximum. The lower bound must be a floor, not a step: a bare
+    ///     <c>subtract-then-guard-positive</c> collapses the budget to a handful of bytes just above
+    ///     the reserve (e.g. at the validator-permitted floor MaxMessageBytes = 1024 + 64 KiB the
+    ///     subtraction leaves 1024, too small to move even one byte-heavy event, so every drain
+    ///     truncates and the drain-until-empty caller never terminates). Taking the max with
+    ///     half the negotiated maximum keeps the budget monotonic across the reserve boundary while
+    ///     still leaving the full reserve for the wrapper whenever the frame max is large enough
+    ///     that the reserve is the smaller subtraction — which is every configuration above 128 KiB.
     /// </summary>
     private int ResolveDrainReplyByteBudget()
     {
-        int budget = _options.MaxMessageBytes - DrainReplyFrameHeadroomBytes;
-        return budget > 0 ? budget : _options.MaxMessageBytes / 2;
+        return Math.Max(
+            _options.MaxMessageBytes - DrainReplyFrameHeadroomBytes,
+            _options.MaxMessageBytes / 2);
     }
 
     private static string CreateDrainTruncationMessage(
@@ -826,11 +867,12 @@ public sealed class WorkerPipeSession
             {
                 // An oversized STA command reply is a property of that one command, not of the
                 // session. Answer the correlation with an error reply instead of falling into the
-                // generic catch below, which would fault the whole session for it.
+                // generic catch below, which would fault the whole session for it. The fallback
+                // write is itself size-guarded (see WriteReplyTooLargeFallbackAsync) so a degenerate
+                // negotiated maximum cannot make even this backstop session-fatal.
                 LogControlReplyTooLarge(envelope.CorrelationId, command.Kind, sizeException);
-                await WriteControlReplyAsync(
-                    CreateReplyTooLargeReply(envelope.CorrelationId, command.Kind),
-                    cancellationToken).ConfigureAwait(false);
+                await WriteReplyTooLargeFallbackAsync(envelope.CorrelationId, command.Kind, cancellationToken)
+                    .ConfigureAwait(false);
             }
         }
         catch (Exception exception) when (exception is not OperationCanceledException)

src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs

@@ -26,9 +26,13 @@ public sealed class MxAccessEventQueue
     /// </summary>
     public const int DefaultCapacity = 10000;
 
-    // Per-event allowance added to WorkerEvent.CalculateSize() when charging the byte budget in
-    // Drain(maxEvents, maxTotalBytes): conservatively covers the repeated-field tag byte and the
-    // length varint the event costs once packed into DrainEventsReply.
+    // Extra per-event slack added to WorkerEvent.CalculateSize() when charging the byte budget in
+    // Drain(maxEvents, maxTotalBytes). CalculateSize() already accounts for the event's own tag and
+    // length-delimiter (the WorkerEvent wrapper serializes MxEvent as field 1, and the reply packs
+    // each MxEvent as DrainEventsReply.events field 1 with the identical tag+length shape), so this
+    // is a pure safety margin over an already-conservative estimate — not compensation for a missing
+    // wrapper. It keeps the running total strictly ahead of the true serialized size so a rounding
+    // edge can never push the packed reply past the frame maximum.
     private const int RepeatedFieldOverheadBytes = 8;
 
     private readonly int capacity;
@@ -221,10 +225,11 @@ public sealed class MxAccessEventQueue
     /// <remarks>
     ///     The size decision happens inside the queue lock, so an event is dequeued only once it is
     ///     known to fit: an event that does not fit stays at the head for the next call and is never
-    ///     lost (WRK-21). Per-event cost is <c>WorkerEvent.CalculateSize()</c> plus
-    ///     <see cref="RepeatedFieldOverheadBytes"/>; the <see cref="WorkerEvent"/> wrapper slightly
-    ///     overestimates the packed <c>MxEvent</c> and the constant conservatively covers the
-    ///     repeated-field tag and length varint, so the estimate errs strictly on the safe side.
+    ///     lost (WRK-21). Per-event cost is <c>WorkerEvent.CalculateSize()</c> — which already
+    ///     includes the event's own tag and length prefix, the same shape the reply's
+    ///     <c>events</c> repeated field packs it into — plus <see cref="RepeatedFieldOverheadBytes"/>
+    ///     of pure slack, so the running total stays strictly ahead of the true serialized size and
+    ///     the estimate errs on the safe side.
     /// </remarks>
     /// <param name="maxEvents">Maximum number of events to drain; 0 means "no count limit".</param>
     /// <param name="maxTotalBytes">Byte budget for the drained events' estimated serialized size.</param>