fix(WRK-21,WRK-28,WRK-23,IPC-30): byte-budget the DrainEvents reply, stop size errors from killing sessions

WRK-21 — DrainEvents was bounded by event count only, so a byte-heavy queue
(large string/array MxValues) built a reply above the negotiated frame maximum:
the writer rejected the frame, the exception unwound the session, and the events
already dequeued were destroyed. The drain is now byte-budgeted inside the queue
lock, so an event is dequeued only once it is known to fit and one that does not
stays at the head. Truncation is reported through the reply's existing
DiagnosticMessage (no contract change); callers drain until an empty reply. Both
reply-write seams — the control-command path and ProcessCommandAsync — now catch
MessageTooLarge and answer the correlation with an InvalidRequest reply instead
of unwinding or faulting the session. Satisfies IPC-23 R1-R3.

WRK-28 — the 10,000 drain ceiling moves to GatewayContractInfo
.MaxDrainEventsPerCommand, referenced by both the gateway request validator and
the worker clamp, replacing a comment-only sync contract. C# const only; no
.proto change.

WRK-23 — WorkerFrameWriter now peek-stamps, validates, then commits the sequence
counter immediately before the stream write, so a per-frame rejection leaves no
phantom gap on the wire.

IPC-30 — an oversized event frame stays session-fatal (it is undeliverable end to
end and neither dropping nor synthesizing a replacement is allowed), but the
death is structured: the event's identity and sizes are logged (never its value),
a WorkerFault with category PROTOCOL_VIOLATION and command method EventDrain is
written, then the session exits as before.

Docs updated in the same change: MxAccessWorkerInstanceDesign.md (drain byte cap,
truncation contract, oversized-head behavior, oversized-event policy, no control
reply is session-fatal on size), WorkerFrameProtocol.md (reply pre-sizing,
non-fatal reply-size rule, oversized-event policy, rejected frames do not consume
sequence numbers), gateway.md (DrainEvents two-axis bound).

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

@@ -19,6 +19,14 @@ public sealed class WorkerPipeSessionTests
     private const string SessionId = "session-1";
     private const string Nonce = "nonce-secret";
 
+    // Byte-heavy drain fixture (WRK-21). 10,000 events at ~1.7 KiB each is ~17 MB of queue — far
+    // more than one frame — so DrainEvents must split across replies. The negotiated frame maximum
+    // is deliberately smaller than the compile-time default so the split happens in a handful of
+    // multi-MB frames instead of moving 17 MB through the test pipe.
+    private const int ByteHeavyEventCount = 10_000;
+    private const int ByteHeavyEventPayloadBytes = 1_800;
+    private const uint NegotiatedMaxFrameBytes = 2 * 1024 * 1024;
+
     /// <summary>Verifies that valid gateway hello triggers worker hello and ready responses.</summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
     [Fact]
@@ -487,6 +495,291 @@ public sealed class WorkerPipeSessionTests
         await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
     }
 
+    /// <summary>
+    ///     The WRK-21 repro. A queue full of byte-heavy events (large string values — the payload
+    ///     profile this gateway exists for) used to make <c>DrainEvents max_events = 0</c> build a
+    ///     reply above the negotiated frame maximum: the writer rejected the frame, the exception
+    ///     unwound the session, and the already-dequeued events were gone. The drain is now
+    ///     byte-budgeted, so the reply fits, the truncation is reported in the reply's diagnostic
+    ///     message, and the session keeps serving.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DrainEvents_ByteHeavyQueue_ReplyIsBoundedAndSessionSurvives()
+    {
+        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(60));
+        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+        FakeRuntimeSession runtime = new()
+        {
+            SuppressDrainForBatchSize = 128,
+            BackingQueue = CreateByteHeavyQueue(ByteHeavyEventCount, ByteHeavyEventPayloadBytes),
+        };
+        WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime);
+        Task runTask = session.RunAsync(cancellation.Token);
+        await CompleteGatewayHandshakeAsync(pipePair, NegotiatedMaxFrameBytes, cancellation.Token);
+
+        await pipePair.GatewayWriter
+            .WriteAsync(
+                CreateControlCommandEnvelope(
+                    "drain-heavy-1",
+                    MxCommandKind.DrainEvents,
+                    command => command.DrainEvents = new DrainEventsCommand { MaxEvents = 0 }),
+                cancellation.Token);
+
+        WorkerEnvelope replyEnvelope = await ReadUntilAsync(
+            pipePair.GatewayReader,
+            WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
+            cancellation.Token);
+
+        MxCommandReply reply = replyEnvelope.WorkerCommandReply.Reply;
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+
+        // The whole queue is far larger than one frame, so the reply is a strict subset that fits.
+        Assert.True(
+            replyEnvelope.CalculateSize() <= NegotiatedMaxFrameBytes,
+            $"DrainEvents reply serialized to {replyEnvelope.CalculateSize()} bytes, above the negotiated {NegotiatedMaxFrameBytes}.");
+        Assert.InRange(reply.DrainEvents.Events.Count, 1, ByteHeavyEventCount - 1);
+        Assert.Contains("remain", reply.DiagnosticMessage);
+        Assert.Contains("repeat DrainEvents", reply.DiagnosticMessage);
+
+        // The session is alive: it still answers a ping, and RunAsync has not unwound.
+        await pipePair.GatewayWriter
+            .WriteAsync(CreatePingCommandEnvelope("ping-after-drain", "still-here"), cancellation.Token);
+        WorkerEnvelope pingReply = await ReadUntilAsync(
+            pipePair.GatewayReader,
+            WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
+            envelope => envelope.WorkerCommandReply.Reply.CorrelationId == "ping-after-drain",
+            cancellation.Token);
+        Assert.Equal("still-here", pingReply.WorkerCommandReply.Reply.DiagnosticMessage);
+        Assert.False(runTask.IsCompleted, "The session must survive a byte-heavy DrainEvents.");
+
+        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
+    }
+
+    /// <summary>
+    ///     Verifies the byte-budgeted drain loses nothing: repeating DrainEvents until it comes back
+    ///     empty recovers every enqueued event exactly once, in order, across the split replies. The
+    ///     pre-fix drain removed events from the queue before sizing the reply, so a rejected frame
+    ///     destroyed them — no-loss is the half of the P0 criterion a catch-only fix cannot deliver.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DrainEvents_RepeatedCalls_RecoverAllEventsWithoutLoss()
+    {
+        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(90));
+        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+        FakeRuntimeSession runtime = new()
+        {
+            SuppressDrainForBatchSize = 128,
+            BackingQueue = CreateByteHeavyQueue(ByteHeavyEventCount, ByteHeavyEventPayloadBytes),
+        };
+        WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime);
+        Task runTask = session.RunAsync(cancellation.Token);
+        await CompleteGatewayHandshakeAsync(pipePair, NegotiatedMaxFrameBytes, cancellation.Token);
+
+        List<ulong> recovered = new();
+        int replyCount = 0;
+        while (true)
+        {
+            string correlationId = $"drain-loop-{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() <= NegotiatedMaxFrameBytes,
+                $"DrainEvents reply {replyCount} serialized to {replyEnvelope.CalculateSize()} bytes.");
+            if (reply.DrainEvents.Events.Count == 0)
+            {
+                break;
+            }
+
+            foreach (MxEvent drained in reply.DrainEvents.Events)
+            {
+                recovered.Add(drained.WorkerSequence);
+            }
+
+            Assert.True(replyCount < 100, "DrainEvents made no progress across 100 replies.");
+        }
+
+        // More than one reply proves the drain really split; every event came back exactly once, in
+        // enqueue order.
+        Assert.True(replyCount > 2, $"Expected the byte cap to split the drain, saw {replyCount} replies.");
+        Assert.Equal(ByteHeavyEventCount, recovered.Count);
+        for (int index = 0; index < recovered.Count; index++)
+        {
+            Assert.Equal((ulong)(index + 1), recovered[index]);
+        }
+
+        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
+    ///     to reach — but that is the point: a future command or a sizing bug must degrade to an
+    ///     error reply for that correlation, never to a dead session.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ControlReplyTooLarge_WritesErrorReplyInsteadOfDying()
+    {
+        const uint tinyMaxFrameBytes = 4096;
+        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15));
+        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+        FakeRuntimeSession runtime = new()
+        {
+            SuppressDrainForBatchSize = 128,
+            BackingQueue = CreateByteHeavyQueue(eventCount: 1, payloadBytes: 16 * 1024),
+            IgnoreDrainByteBudget = true,
+        };
+        WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime);
+        Task runTask = session.RunAsync(cancellation.Token);
+        await CompleteGatewayHandshakeAsync(pipePair, tinyMaxFrameBytes, cancellation.Token);
+
+        await pipePair.GatewayWriter
+            .WriteAsync(
+                CreateControlCommandEnvelope(
+                    "drain-too-large",
+                    MxCommandKind.DrainEvents,
+                    command => command.DrainEvents = new DrainEventsCommand { MaxEvents = 0 }),
+                cancellation.Token);
+
+        WorkerEnvelope replyEnvelope = await ReadUntilAsync(
+            pipePair.GatewayReader,
+            WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
+            cancellation.Token);
+
+        MxCommandReply reply = replyEnvelope.WorkerCommandReply.Reply;
+        Assert.Equal("drain-too-large", reply.CorrelationId);
+        Assert.Equal(MxCommandKind.DrainEvents, reply.Kind);
+        Assert.Equal(ProtocolStatusCode.InvalidRequest, reply.ProtocolStatus.Code);
+        Assert.Contains("frame maximum", reply.ProtocolStatus.Message);
+        Assert.False(runTask.IsCompleted, "An oversized control reply must not end the session.");
+
+        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
+    }
+
+    /// <summary>
+    ///     Verifies the same backstop on the STA command path: an oversized reply from a dispatched
+    ///     command answers its correlation with an error reply instead of falling into the generic
+    ///     catch that faults the whole session with MxaccessCommandFailed.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task CommandReplyTooLarge_WritesErrorReplyInsteadOfFaulting()
+    {
+        const uint tinyMaxFrameBytes = 4096;
+        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15));
+        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+        FakeRuntimeSession runtime = new()
+        {
+            DispatchReplyDiagnosticMessage = new string('x', 16 * 1024),
+        };
+        WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime);
+        Task runTask = session.RunAsync(cancellation.Token);
+        await CompleteGatewayHandshakeAsync(pipePair, tinyMaxFrameBytes, cancellation.Token);
+
+        await pipePair.GatewayWriter
+            .WriteAsync(CreateCommandEnvelope("command-too-large"), cancellation.Token);
+
+        WorkerEnvelope replyEnvelope = await ReadUntilAsync(
+            pipePair.GatewayReader,
+            WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
+            envelope => envelope.WorkerCommandReply.Reply.CorrelationId == "command-too-large",
+            cancellation.Token);
+
+        MxCommandReply reply = replyEnvelope.WorkerCommandReply.Reply;
+        Assert.Equal(MxCommandKind.Register, reply.Kind);
+        Assert.Equal(ProtocolStatusCode.InvalidRequest, reply.ProtocolStatus.Code);
+
+        // No fault, and the session still reports itself Ready rather than Faulted.
+        await pipePair.GatewayWriter
+            .WriteAsync(
+                CreateControlCommandEnvelope(
+                    "state-after-too-large",
+                    MxCommandKind.GetSessionState,
+                    command => command.GetSessionState = new GetSessionStateCommand()),
+                cancellation.Token);
+        WorkerEnvelope stateEnvelope = await ReadUntilAsync(
+            pipePair.GatewayReader,
+            WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
+            envelope => envelope.WorkerCommandReply.Reply.CorrelationId == "state-after-too-large",
+            cancellation.Token);
+        Assert.Equal(SessionState.Ready, stateEnvelope.WorkerCommandReply.Reply.SessionState.State);
+        Assert.False(runTask.IsCompleted, "An oversized command reply must not fault the session.");
+
+        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
+    }
+
+    /// <summary>
+    ///     IPC-30. An event above the negotiated frame maximum is undeliverable end to end (the pipe
+    ///     maximum sits only an envelope reserve above the public gRPC cap), so the session stays
+    ///     fatal by design — but the death must be structured: a WorkerFault naming the event, with
+    ///     no value payload in it, before the process exits. Silently dropping the event or
+    ///     synthesizing a placeholder were both rejected.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RunAsync_EventFrameTooLarge_WritesStructuredFaultThenEndsSession()
+    {
+        const uint tinyMaxFrameBytes = 4096;
+        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15));
+        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+        FakeRuntimeSession runtime = new();
+        RecordingWorkerLogger logger = new();
+        WorkerPipeSession session = CreatePipeSession(
+            pipePair.WorkerStream,
+            runtime,
+            new WorkerPipeSessionOptions
+            {
+                HeartbeatInterval = TimeSpan.FromMilliseconds(100),
+                HeartbeatGrace = TimeSpan.FromSeconds(5),
+            },
+            logger);
+        runtime.EnqueueEvent(CreateOversizedWorkerEvent(sequence: 77, payloadBytes: 16 * 1024));
+        Task runTask = session.RunAsync(cancellation.Token);
+        await CompleteGatewayHandshakeAsync(pipePair, tinyMaxFrameBytes, cancellation.Token);
+
+        WorkerEnvelope faultEnvelope = await ReadUntilAsync(
+            pipePair.GatewayReader,
+            WorkerEnvelope.BodyOneofCase.WorkerFault,
+            cancellation.Token);
+
+        WorkerFault fault = faultEnvelope.WorkerFault;
+        Assert.Equal(WorkerFaultCategory.ProtocolViolation, fault.Category);
+        Assert.Equal("EventDrain", fault.CommandMethod);
+        Assert.Contains("77", fault.DiagnosticMessage);
+        Assert.Contains("MaxMessageBytes", fault.DiagnosticMessage);
+        // The identity is reported; the value payload never is.
+        Assert.DoesNotContain(new string('x', 64), fault.DiagnosticMessage);
+
+        Assert.Contains(
+            logger.Events,
+            entry => entry.EventName == "WorkerEventFrameTooLarge"
+                && entry.Fields.TryGetValue("worker_sequence", out object? sequence)
+                && sequence is ulong sequenceValue
+                && sequenceValue == 77UL);
+
+        // The session ends, and the fault frame parsed cleanly off the same stream above — the
+        // rejected event never corrupted the wire.
+        Task completedTask = await Task.WhenAny(runTask, Task.Delay(TimeSpan.FromSeconds(5), cancellation.Token));
+        Assert.Same(runTask, completedTask);
+        await Assert.ThrowsAsync<InvalidOperationException>(async () => await runTask);
+    }
+
     /// <summary>
     ///     Verifies that ShutdownWorker returns its OK reply BEFORE the graceful
     ///     shutdown runs and disposes the runtime session, and that the message
@@ -1241,6 +1534,15 @@ public sealed class WorkerPipeSessionTests
         Stream stream,
         FakeRuntimeSession runtime,
         WorkerPipeSessionOptions sessionOptions)
+    {
+        return CreatePipeSession(stream, runtime, sessionOptions, logger: null);
+    }
+
+    private static WorkerPipeSession CreatePipeSession(
+        Stream stream,
+        FakeRuntimeSession runtime,
+        WorkerPipeSessionOptions sessionOptions,
+        ZB.MOM.WW.MxGateway.Worker.Bootstrap.IWorkerLogger? logger)
     {
         WorkerFrameProtocolOptions options = CreateOptions();
         return new WorkerPipeSession(
@@ -1249,7 +1551,8 @@ public sealed class WorkerPipeSessionTests
             options,
             () => 1234,
             sessionOptions,
-            () => runtime);
+            () => runtime,
+            logger);
     }
 
     private static WorkerFrameProtocolOptions CreateOptions()
@@ -1270,7 +1573,8 @@ public sealed class WorkerPipeSessionTests
     private static WorkerEnvelope CreateGatewayHelloEnvelope(
         string nonce = Nonce,
         uint supportedProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
-        ulong sequence = 1)
+        ulong sequence = 1,
+        uint maxFrameBytes = 0)
     {
         return new WorkerEnvelope
         {
@@ -1282,6 +1586,10 @@ public sealed class WorkerPipeSessionTests
                 SupportedProtocolVersion = supportedProtocolVersion,
                 Nonce = nonce,
                 GatewayVersion = "test-gateway",
+                // 0 leaves the worker on its compile-time default; a non-zero value is adopted
+                // during the handshake and becomes the frame maximum every later assertion is
+                // measured against.
+                MaxFrameBytes = maxFrameBytes,
             },
         };
     }
@@ -1392,12 +1700,55 @@ public sealed class WorkerPipeSessionTests
         };
     }
 
-    private static async Task CompleteGatewayHandshakeAsync(
+    private static WorkerEvent CreateOversizedWorkerEvent(ulong sequence, int payloadBytes)
+    {
+        WorkerEvent workerEvent = CreateWorkerEvent(sequence);
+        workerEvent.Event.ItemHandle = 42;
+        workerEvent.Event.RawStatus = new string('x', payloadBytes);
+        return workerEvent;
+    }
+
+    /// <summary>
+    /// Fills a real event queue with byte-heavy events — a large string field stands in for the
+    /// array/string <c>MxValue</c> payloads that make a count-capped drain overshoot the frame
+    /// maximum. The queue is real (not the fake's plain list) so the production byte-budgeting runs.
+    /// </summary>
+    /// <param name="eventCount">Number of events to enqueue.</param>
+    /// <param name="payloadBytes">Size of each event's raw-status payload string.</param>
+    /// <returns>The populated queue.</returns>
+    private static MxAccessEventQueue CreateByteHeavyQueue(int eventCount, int payloadBytes)
+    {
+        MxAccessEventQueue queue = new(Math.Max(eventCount, 1));
+        string payload = new string('x', payloadBytes);
+        for (int index = 0; index < eventCount; index++)
+        {
+            queue.Enqueue(new MxEvent
+            {
+                SessionId = SessionId,
+                Family = MxEventFamily.OnDataChange,
+                ItemHandle = index,
+                RawStatus = payload,
+                OnDataChange = new OnDataChangeEvent(),
+            });
+        }
+
+        return queue;
+    }
+
+    private static Task CompleteGatewayHandshakeAsync(
         PipePair pipePair,
         CancellationToken cancellationToken)
+    {
+        return CompleteGatewayHandshakeAsync(pipePair, maxFrameBytes: 0, cancellationToken);
+    }
+
+    private static async Task CompleteGatewayHandshakeAsync(
+        PipePair pipePair,
+        uint maxFrameBytes,
+        CancellationToken cancellationToken)
     {
         await pipePair.GatewayWriter
-            .WriteAsync(CreateGatewayHelloEnvelope(), cancellationToken)
+            .WriteAsync(CreateGatewayHelloEnvelope(maxFrameBytes: maxFrameBytes), cancellationToken)
             .ConfigureAwait(false);
 
         WorkerEnvelope hello = await pipePair.GatewayReader.ReadAsync(cancellationToken).ConfigureAwait(false);