Merge remote-tracking branch 'origin/fix/wrk-21-drain-cluster'

# Conflicts:
#	archreview/2026-07-12/remediation/00-tracking.md
#	docs/MxAccessWorkerInstanceDesign.md

src/ZB.MOM.WW.MxGateway.Contracts/GatewayContractInfo.cs

@@ -17,6 +17,21 @@ public static class GatewayContractInfo
     /// <summary>Default backend name identifying the MXAccess worker process type.</summary>
     public const string DefaultBackendName = "mxaccess-worker";
 
+    /// <summary>
+    /// Ceiling on how many events one <c>DrainEvents</c> command may move in a single reply.
+    /// Shared so the gateway's request-validation ceiling
+    /// (<c>MxAccessGrpcRequestValidator</c>, which rejects a larger <c>max_events</c> loudly at
+    /// the public boundary) and the worker's per-reply clamp
+    /// (<c>WorkerPipeSession.CreateDrainEventsReply</c>, the backstop that also interprets
+    /// <c>max_events = 0</c>) cannot drift apart. A count cap alone is necessary but not
+    /// sufficient: the worker additionally caps the reply by serialized bytes against the
+    /// negotiated frame maximum (WRK-21), so a reply may carry fewer events than this ceiling
+    /// and fewer than are queued. Callers drain iteratively until an empty reply.
+    /// This is a documented behavioral bound, not wire schema — it is deliberately a C#
+    /// constant and not a <c>.proto</c> field.
+    /// </summary>
+    public const uint MaxDrainEventsPerCommand = 10_000;
+
     /// <summary>
     /// Environment variable name that opts an xUnit suite into running live
     /// MXAccess COM tests. Single source of truth shared by both

src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs

@@ -1,17 +1,11 @@
 using Grpc.Core;
+using ZB.MOM.WW.MxGateway.Contracts;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
 
 namespace ZB.MOM.WW.MxGateway.Server.Grpc;
 
 public sealed class MxAccessGrpcRequestValidator
 {
-    // Upper bound on a single DrainEvents request. DrainEvents is a diagnostics RPC that returns
-    // buffered events in one non-streaming reply, so an unbounded max_events could pack the whole
-    // queue into a session-killing frame. The worker independently caps each reply at its
-    // own MaxDrainEventsPerReply; this public bound rejects an obviously-abusive request loudly at
-    // the boundary. max_events = 0 is allowed and means "the worker's default batch cap".
-    private const uint MaxDrainEventsPerRequest = 10_000;
-
     /// <summary>Validates an open session request.</summary>
     /// <param name="request">The request to validate.</param>
     public void ValidateOpenSession(OpenSessionRequest request)
@@ -78,10 +72,18 @@ public sealed class MxAccessGrpcRequestValidator
         }
 
         // The payload case now matches the kind, so command.DrainEvents is non-null here.
-        if (command.Kind is MxCommandKind.DrainEvents && command.DrainEvents.MaxEvents > MaxDrainEventsPerRequest)
+        // DrainEvents is a diagnostics RPC that returns buffered events in one non-streaming
+        // reply, so an unbounded max_events could pack the whole queue into a session-killing
+        // frame. The worker independently clamps every reply to the same shared ceiling and
+        // additionally caps it by serialized bytes; this public bound rejects an obviously-abusive
+        // request loudly at the boundary. max_events = 0 is allowed and means "the worker's
+        // default batch cap".
+        if (command.Kind is MxCommandKind.DrainEvents
+            && command.DrainEvents.MaxEvents > GatewayContractInfo.MaxDrainEventsPerCommand)
         {
             throw InvalidArgument(
-                $"DrainEvents max_events ({command.DrainEvents.MaxEvents}) must not exceed {MaxDrainEventsPerRequest}; "
+                $"DrainEvents max_events ({command.DrainEvents.MaxEvents}) must not exceed "
+                + $"{GatewayContractInfo.MaxDrainEventsPerCommand}; "
                 + "use 0 to request the worker default batch cap.");
         }
     }

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

@@ -410,6 +410,48 @@ public sealed class WorkerFrameProtocolTests
         }
     }
 
+    /// <summary>
+    /// Verifies a per-frame rejection does not burn a sequence number (WRK-23). The sequence is a
+    /// diagnostic counter, so a gap breaks nothing functionally — but an operator correlating a pipe
+    /// capture reads a gap as a lost frame and chases a bug that does not exist, and the gap-free
+    /// guarantee the concurrent-write test asserts would otherwise only hold until the first
+    /// rejection.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WriteAsync_PerFrameRejection_DoesNotConsumeSequence()
+    {
+        const int maxMessageBytes = 512;
+        WorkerFrameProtocolOptions options = new(
+            SessionId,
+            GatewayContractInfo.WorkerProtocolVersion,
+            Nonce,
+            maxMessageBytes);
+        using MemoryStream stream = new();
+        WorkerFrameWriter writer = new(stream, options);
+
+        await writer.WriteAsync(CreateEventEnvelope());
+
+        WorkerEnvelope oversized = CreateGatewayHelloEnvelope();
+        oversized.GatewayHello.GatewayVersion = new string('x', maxMessageBytes * 2);
+        WorkerFrameProtocolException exception =
+            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
+                async () => await writer.WriteAsync(oversized));
+        Assert.Equal(WorkerFrameProtocolErrorCode.MessageTooLarge, exception.ErrorCode);
+
+        await writer.WriteAsync(CreateEventEnvelope());
+
+        stream.Position = 0;
+        WorkerFrameReader reader = new(stream, options);
+        WorkerEnvelope first = await reader.ReadAsync();
+        WorkerEnvelope second = await reader.ReadAsync();
+
+        // Two frames reached the wire; the rejected frame in between left no gap.
+        Assert.Equal(1UL, first.Sequence);
+        Assert.Equal(2UL, second.Sequence);
+        Assert.Equal(stream.Length, stream.Position);
+    }
+
     /// <summary>Verifies a zero negotiated frame maximum keeps the constructor default.</summary>
     [Fact]
     public void AdoptNegotiatedMaxMessageBytes_WithZero_KeepsDefault()

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

@@ -19,6 +19,21 @@ 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 truncate.
+    //
+    // Two limits below are harness accommodations, not properties of the fix. PipePair runs both
+    // ends of a duplex pipe inside one process, with no continuous read pump and with blocking
+    // FlushFileBuffers under every frame write, so it tolerates neither multi-megabyte frames nor
+    // hundreds of large round trips before both ends wedge waiting on each other. Hence a small
+    // negotiated frame maximum, and a smaller queue for the drain-to-empty walk. The byte cap
+    // behaves identically at any frame size; exhaustive no-loss over the full 10,000 events is
+    // covered without a pipe by MxAccessEventQueueTests.
+    private const int ByteHeavyEventCount = 10_000;
+    private const int RepeatedDrainEventCount = 1_000;
+    private const int ByteHeavyEventPayloadBytes = 1_800;
+    private const uint NegotiatedMaxFrameBytes = 128 * 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 +502,361 @@ 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(RepeatedDrainEventCount, 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 < 200, "DrainEvents made no progress across 200 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(RepeatedDrainEventCount, recovered.Count);
+        for (int index = 0; index < recovered.Count; index++)
+        {
+            Assert.Equal((ulong)(index + 1), recovered[index]);
+        }
+
+        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
+    ///     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 +1611,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 +1628,8 @@ public sealed class WorkerPipeSessionTests
             options,
             () => 1234,
             sessionOptions,
-            () => runtime);
+            () => runtime,
+            logger);
     }
 
     private static WorkerFrameProtocolOptions CreateOptions()
@@ -1270,7 +1650,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 +1663,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 +1777,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);

src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs

@@ -98,6 +98,157 @@ public sealed class MxAccessEventQueueTests
         Assert.Equal(0, queue.Count);
     }
 
+    /// <summary>
+    ///     Verifies the byte-budgeted drain stops before the budget is exceeded, leaves the
+    ///     remainder queued in order, and reports the exact remaining count (WRK-21). Events that
+    ///     do not fit must never be dequeued — dequeuing them is how the pre-fix drain lost events
+    ///     when the reply frame was rejected.
+    /// </summary>
+    [Fact]
+    public void Drain_ByteBudget_StopsBeforeBudgetAndLeavesRemainderQueued()
+    {
+        MxAccessEventQueue queue = new(capacity: 8);
+        for (int itemHandle = 0; itemHandle < 5; itemHandle++)
+        {
+            queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 512));
+        }
+
+        int perEventCost = MeasureDrainCost(payloadLength: 512);
+
+        // Budget for exactly two events (plus a sliver too small for a third).
+        IReadOnlyList<WorkerEvent> drained =
+            queue.Drain(maxEvents: 0, maxTotalBytes: (perEventCost * 2) + (perEventCost / 2)).Events;
+
+        Assert.Equal(2, drained.Count);
+        Assert.Equal(0, drained[0].Event.ItemHandle);
+        Assert.Equal(1, drained[1].Event.ItemHandle);
+        Assert.Equal(3, queue.Count);
+
+        // The undrained remainder is still present, still in order.
+        IReadOnlyList<WorkerEvent> rest = queue.Drain(maxEvents: 0);
+        Assert.Equal(new[] { 2, 3, 4 }, new[] { rest[0].Event.ItemHandle, rest[1].Event.ItemHandle, rest[2].Event.ItemHandle });
+    }
+
+    /// <summary>
+    ///     Verifies the byte-budgeted drain reports truncation and the exact remaining count so the
+    ///     DrainEvents reply can tell the caller to drain again.
+    /// </summary>
+    [Fact]
+    public void Drain_ByteBudget_ReportsTruncationAndRemainingCount()
+    {
+        MxAccessEventQueue queue = new(capacity: 8);
+        for (int itemHandle = 0; itemHandle < 4; itemHandle++)
+        {
+            queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 256));
+        }
+
+        // One-and-a-half events' worth of budget: the head fits, the next does not, and the next is
+        // comfortably smaller than the whole budget so it is a plain truncation rather than the
+        // oversized-head case.
+        WorkerEventDrainResult result = queue.Drain(
+            maxEvents: 0,
+            maxTotalBytes: MeasureDrainCost(payloadLength: 256) * 3 / 2);
+
+        Assert.Single(result.Events);
+        Assert.True(result.TruncatedBySize);
+        Assert.Equal(3, result.RemainingCount);
+        Assert.Equal(0UL, result.OversizedHeadSequence);
+    }
+
+    /// <summary>
+    ///     Verifies the degenerate case: a head event whose own serialized size exceeds the whole
+    ///     budget is not drained (draining it would build an oversized reply or lose the event) and
+    ///     its worker sequence is reported so an operator can find the offending tag.
+    /// </summary>
+    [Fact]
+    public void Drain_ByteBudget_OversizedHead_DrainsNothingAndReportsHeadSequence()
+    {
+        MxAccessEventQueue queue = new(capacity: 8);
+        queue.Enqueue(CreateEventWithPayload(itemHandle: 0, payloadLength: 4096));
+        queue.Enqueue(CreateEventWithPayload(itemHandle: 1, payloadLength: 8));
+
+        WorkerEventDrainResult result = queue.Drain(maxEvents: 0, maxTotalBytes: 1024);
+
+        Assert.Empty(result.Events);
+        Assert.True(result.TruncatedBySize);
+        Assert.Equal(2, result.RemainingCount);
+        Assert.Equal(1UL, result.OversizedHeadSequence);
+
+        // The blocked event is still queued — it was never removed.
+        Assert.Equal(2, queue.Count);
+    }
+
+    /// <summary>
+    ///     The no-loss half of the WRK-21 acceptance criterion, at full scale. Draining the review's
+    ///     10,000 byte-heavy events under a budget that fits only a fraction of them per call must
+    ///     return every event exactly once and in order: the pre-fix drain removed events from the
+    ///     queue before the reply was sized, so a rejected frame destroyed them. This runs at the
+    ///     queue layer because the property is the queue's, and because the pipe harness that covers
+    ///     the same walk end to end cannot sustain hundreds of large round trips.
+    /// </summary>
+    [Fact]
+    public void Drain_ByteBudget_RepeatedCalls_RecoverAllEventsInOrderWithoutLoss()
+    {
+        const int eventCount = 10_000;
+        const int payloadLength = 1_800;
+        MxAccessEventQueue queue = new(eventCount);
+        for (int index = 0; index < eventCount; index++)
+        {
+            queue.Enqueue(CreateEventWithPayload(index, payloadLength));
+        }
+
+        // A budget that fits roughly 35 events, so the walk takes hundreds of calls.
+        int budget = MeasureDrainCost(payloadLength) * 35;
+        List<ulong> recovered = new();
+        int calls = 0;
+        while (true)
+        {
+            WorkerEventDrainResult result = queue.Drain(maxEvents: 0, maxTotalBytes: budget);
+            calls++;
+            if (result.Events.Count == 0)
+            {
+                break;
+            }
+
+            foreach (WorkerEvent drained in result.Events)
+            {
+                recovered.Add(drained.Event.WorkerSequence);
+            }
+
+            Assert.Equal(eventCount - recovered.Count, result.RemainingCount);
+            Assert.True(calls < eventCount, "Drain made no progress.");
+        }
+
+        Assert.True(calls > 100, $"Expected the byte budget to split the drain, saw {calls} calls.");
+        Assert.Equal(eventCount, recovered.Count);
+        for (int index = 0; index < recovered.Count; index++)
+        {
+            Assert.Equal((ulong)(index + 1), recovered[index]);
+        }
+
+        Assert.Equal(0, queue.Count);
+    }
+
+    /// <summary>
+    ///     Verifies the count cap still binds when the byte budget is generous: the byte cap is an
+    ///     additional bound, not a replacement.
+    /// </summary>
+    [Fact]
+    public void Drain_ByteBudget_CountCapStillBinds()
+    {
+        MxAccessEventQueue queue = new(capacity: 8);
+        for (int itemHandle = 0; itemHandle < 5; itemHandle++)
+        {
+            queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 16));
+        }
+
+        WorkerEventDrainResult result = queue.Drain(maxEvents: 2, maxTotalBytes: 1024 * 1024);
+
+        Assert.Equal(2, result.Events.Count);
+        Assert.False(result.TruncatedBySize);
+        Assert.Equal(3, result.RemainingCount);
+    }
+
     /// <summary>Verifies that Enqueue is rejected after a fault is recorded manually.</summary>
     [Fact]
     public void Enqueue_AfterRecordFault_ThrowsInvalidOperationException()
@@ -149,6 +300,42 @@ public sealed class MxAccessEventQueueTests
         Assert.Equal(WorkerFaultCategory.MxaccessEventConversionFailed, queue.Fault?.Category);
     }
 
+    // Mirrors MxAccessEventQueue's per-event repeated-field allowance. Kept local rather than made
+    // public on the queue: the byte-budget tests state their budgets in units of that charge, so a
+    // change to it should surface here as a failing bound instead of silently moving with the code.
+    private const int RepeatedFieldOverheadBytes = 8;
+
+    /// <summary>
+    /// Measures what the queue charges one event of the given payload size against the byte budget:
+    /// the serialized <see cref="WorkerEvent"/> as it exists after Enqueue (sequence and timestamp
+    /// stamped) plus the repeated-field allowance. The probe uses item handle 0, a proto3 default
+    /// that is not serialized, so this is a lower bound on the fixtures' real per-event cost — the
+    /// budgets above carry slack rather than assuming byte equality.
+    /// </summary>
+    /// <param name="payloadLength">Length of the event's raw-status payload string.</param>
+    /// <returns>The per-event byte cost.</returns>
+    private static int MeasureDrainCost(int payloadLength)
+    {
+        MxAccessEventQueue probe = new(capacity: 1);
+        probe.Enqueue(CreateEventWithPayload(0, payloadLength));
+        Assert.True(probe.TryDequeue(out WorkerEvent? probeEvent));
+        return probeEvent!.CalculateSize() + RepeatedFieldOverheadBytes;
+    }
+
+    /// <summary>
+    /// Builds a byte-heavy event: a large string field is the cheapest stand-in for the array/string
+    /// <see cref="MxValue"/> payloads that make a count-capped drain overshoot the frame maximum.
+    /// </summary>
+    /// <param name="itemHandle">Item handle identifying the event in assertions.</param>
+    /// <param name="payloadLength">Length of the raw-status payload string.</param>
+    /// <returns>The constructed event.</returns>
+    private static MxEvent CreateEventWithPayload(int itemHandle, int payloadLength)
+    {
+        MxEvent mxEvent = CreateEvent(MxEventFamily.OnDataChange, itemHandle);
+        mxEvent.RawStatus = new string('x', payloadLength);
+        return mxEvent;
+    }
+
     private static MxEvent CreateEvent(
         MxEventFamily family,
         int itemHandle)

src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs

@@ -43,6 +43,13 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
     /// <summary>Gets or sets whether ShutdownGracefullyAsync throws a TimeoutException.</summary>
     public bool ThrowTimeoutOnShutdown { get; set; }
 
+    /// <summary>
+    ///     Optional diagnostic message stuffed into every dispatched command reply. A long value
+    ///     pushes the STA command reply past a small negotiated frame maximum, which is how a test
+    ///     drives the <c>ProcessCommandAsync</c> reply-size backstop.
+    /// </summary>
+    public string? DispatchReplyDiagnosticMessage { get; set; }
+
     /// <summary>Gets a value indicating whether Dispose was called.</summary>
     public bool Disposed { get; private set; }
 
@@ -92,7 +99,7 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
                     throw new InvalidOperationException("Command failed after shutdown started.");
                 }
 
-                return new MxCommandReply
+                MxCommandReply reply = new()
                 {
                     SessionId = command.SessionId,
                     CorrelationId = command.CorrelationId,
@@ -103,6 +110,13 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
                         Message = "OK",
                     },
                 };
+
+                if (DispatchReplyDiagnosticMessage is not null)
+                {
+                    reply.DiagnosticMessage = DispatchReplyDiagnosticMessage;
+                }
+
+                return reply;
             });
     }
 
@@ -133,6 +147,27 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
     /// </summary>
     public uint? LastDrainMaxEvents { get; private set; }
 
+    /// <summary>
+    ///     Optional real event queue backing the drain paths. When set, both
+    ///     <see cref="DrainEvents(uint)"/> and <see cref="DrainEvents(uint, int)"/> delegate to it
+    ///     so a test can exercise the production byte-budgeting logic behind the fake session.
+    /// </summary>
+    public MxAccessEventQueue? BackingQueue { get; set; }
+
+    /// <summary>
+    ///     When set, <see cref="DrainEvents(uint, int)"/> ignores the byte budget and drains purely
+    ///     by count. Simulates the "sizing bug or future command" case the control-reply size
+    ///     backstop exists for, so a test can drive an oversized reply without a real budgeting
+    ///     defect.
+    /// </summary>
+    public bool IgnoreDrainByteBudget { get; set; }
+
+    /// <summary>
+    ///     Records the <c>maxTotalBytes</c> argument of the most recent byte-budgeted
+    ///     <see cref="DrainEvents(uint, int)"/> call.
+    /// </summary>
+    public int? LastDrainMaxTotalBytes { get; private set; }
+
     /// <inheritdoc />
     public IReadOnlyList<WorkerEvent> DrainEvents(uint maxEvents)
     {
@@ -143,6 +178,76 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
 
         LastDrainMaxEvents = maxEvents;
 
+        if (BackingQueue is not null)
+        {
+            return BackingQueue.Drain(maxEvents);
+        }
+
+        lock (gate)
+        {
+            int drainCount = maxEvents == 0
+                ? events.Count
+                : Math.Min(events.Count, checked((int)Math.Min(maxEvents, int.MaxValue)));
+            List<WorkerEvent> drained = new(drainCount);
+            for (int index = 0; index < drainCount; index++)
+            {
+                drained.Add(events.Dequeue());
+            }
+
+            return drained;
+        }
+    }
+
+    /// <inheritdoc />
+    public WorkerEventDrainResult DrainEvents(uint maxEvents, int maxTotalBytes)
+    {
+        if (SuppressDrainForBatchSize is uint suppressed && maxEvents == suppressed)
+        {
+            return new WorkerEventDrainResult(
+                Array.Empty<WorkerEvent>(),
+                truncatedBySize: false,
+                remainingCount: PendingEventCount,
+                oversizedHeadSequence: 0);
+        }
+
+        LastDrainMaxEvents = maxEvents;
+        LastDrainMaxTotalBytes = maxTotalBytes;
+
+        if (BackingQueue is not null && !IgnoreDrainByteBudget)
+        {
+            return BackingQueue.Drain(maxEvents, maxTotalBytes);
+        }
+
+        // Count-only drain: either no backing queue (the simple fakes) or a deliberately
+        // budget-blind drain used to exercise the reply-size backstop.
+        IReadOnlyList<WorkerEvent> drained = BackingQueue is not null
+            ? BackingQueue.Drain(maxEvents)
+            : DrainByCount(maxEvents);
+        return new WorkerEventDrainResult(
+            drained,
+            truncatedBySize: false,
+            remainingCount: PendingEventCount,
+            oversizedHeadSequence: 0);
+    }
+
+    private int PendingEventCount
+    {
+        get
+        {
+            if (BackingQueue is not null)
+            {
+                return BackingQueue.Count;
+            }
+
+            lock (gate)
+            {
+                return events.Count;
+            }
+        }
+    }
+
+    private IReadOnlyList<WorkerEvent> DrainByCount(uint maxEvents)
+    {
         lock (gate)
         {
             int drainCount = maxEvents == 0

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

@@ -43,8 +43,9 @@ public sealed class WorkerFrameWriter
     private readonly Queue<PendingFrame> _eventFrames = new Queue<PendingFrame>();
 
     // Only ever read/written by the current write-lock holder while draining, so no interlock is
-    // needed. Starts at 0 and is pre-incremented, so the first written frame carries sequence 1
-    // (matching the previous behaviour).
+    // needed. Starts at 0 and is committed only immediately before the stream write, so the first
+    // written frame carries sequence 1 and a per-frame rejection leaves the counter untouched —
+    // the next accepted frame reuses the number and the wire sequence stays contiguous.
     private ulong _nextSequence;
 
     /// <summary>Initializes a new instance of the WorkerFrameWriter class.</summary>
@@ -237,7 +238,14 @@ public sealed class WorkerFrameWriter
 
         // Stamp the sequence at the actual point of writing, under the write lock, so the wire order
         // and the stamped sequence agree regardless of caller concurrency or priority.
-        envelope.Sequence = unchecked(++_nextSequence);
+        //
+        // Peek-stamp-commit (WRK-23): the sequence participates in CalculateSize() (varint width),
+        // so it must be stamped before the size checks — but a per-frame rejection must not burn a
+        // number, or the wire shows phantom gaps that an operator reads as lost frames. Stamp a
+        // candidate, validate the stamped envelope, and commit the counter only once the frame is
+        // certain to be written.
+        ulong candidateSequence = unchecked(_nextSequence + 1);
+        envelope.Sequence = candidateSequence;
 
         int payloadLength = envelope.CalculateSize();
         if (payloadLength == 0)
@@ -254,6 +262,8 @@ public sealed class WorkerFrameWriter
                 $"Worker envelope payload length {payloadLength} exceeds the configured maximum of {_options.MaxMessageBytes} bytes.");
         }
 
+        _nextSequence = candidateSequence;
+
         // Serialize once into a single buffer that carries the 4-byte length prefix followed by the
         // payload, then issue one stream write. This avoids a second serialization pass, a separate
         // prefix array, and a separate prefix write. The flush is deferred to the end of the drained

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

@@ -5,6 +5,7 @@ using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
 using Google.Protobuf.WellKnownTypes;
+using ZB.MOM.WW.MxGateway.Contracts;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
 using ZB.MOM.WW.MxGateway.Worker.Bootstrap;
 using ZB.MOM.WW.MxGateway.Worker.MxAccess;
@@ -18,12 +19,11 @@ public sealed class WorkerPipeSession
     private static readonly TimeSpan BackgroundTaskStopTimeout = TimeSpan.FromSeconds(1);
     private const uint EventDrainBatchSize = 128;
 
-    // Hard cap on how many events a single DrainEvents diagnostic reply may carry. DrainEvents is a
-    // non-streaming control command, so an unbounded drain (including the max_events = 0 "as many as
-    // available" request) could pack the whole queue into one session-killing reply frame.
-    // The gateway request validator rejects requests above its public ceiling; this worker-side cap is
-    // the backstop and defines the effective per-reply maximum. Kept in step with that public ceiling.
-    private const uint MaxDrainEventsPerReply = 10_000;
+    // Headroom subtracted from the negotiated frame maximum when budgeting a DrainEvents reply. It
+    // covers the WorkerEnvelope/WorkerCommandReply/MxCommandReply wrapper the drained events are
+    // packed into — the same envelope-overhead reserve rationale docs/WorkerFrameProtocol.md
+    // records for the frame max itself.
+    private const int DrainReplyFrameHeadroomBytes = 64 * 1024;
 
     private readonly WorkerFrameProtocolOptions _options;
     private readonly Func<int> _processIdProvider;
@@ -376,13 +376,83 @@ public sealed class WorkerPipeSession
                 // Events are the low-priority frame class: the writer holds them behind any pending
                 // control frame (reply, fault, heartbeat, shutdown ack) so those are not delayed
                 // behind an event backlog.
-                await _writer
-                    .WriteAsync(CreateEnvelope(workerEvent), WorkerFrameWritePriority.Event, cancellationToken)
-                    .ConfigureAwait(false);
+                try
+                {
+                    await _writer
+                        .WriteAsync(CreateEnvelope(workerEvent), WorkerFrameWritePriority.Event, cancellationToken)
+                        .ConfigureAwait(false);
+                }
+                catch (WorkerFrameProtocolException exception)
+                    when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge)
+                {
+                    await FaultOnOversizedEventAsync(workerEvent, exception, cancellationToken)
+                        .ConfigureAwait(false);
+                }
             }
         }
     }
 
+    /// <summary>
+    ///     Ends the session on an event that cannot be framed, but deliberately and diagnosably
+    ///     (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 dropping it
+    ///     would silently make the event stream unfaithful, and synthesizing a placeholder is barred
+    ///     by the no-synthesized-events rule. Instead the worker records which event blocked (never
+    ///     its value: the redaction rule), writes a structured fault the gateway and dashboard can
+    ///     surface, and then exits as it did before. Remediation is configuration:
+    ///     <c>MxGateway:Worker:MaxMessageBytes</c>. Other per-frame rejection codes keep the previous
+    ///     behavior — they indicate worker bugs, not workload size.
+    /// </summary>
+    private async Task FaultOnOversizedEventAsync(
+        WorkerEvent workerEvent,
+        WorkerFrameProtocolException exception,
+        CancellationToken cancellationToken)
+    {
+        MxEvent? mxEvent = workerEvent.Event;
+        string family = (mxEvent?.Family ?? MxEventFamily.Unspecified).ToString();
+        ulong workerSequence = mxEvent?.WorkerSequence ?? 0;
+        int serverHandle = mxEvent?.ServerHandle ?? 0;
+        int itemHandle = mxEvent?.ItemHandle ?? 0;
+
+        _logger?.Error(
+            "WorkerEventFrameTooLarge",
+            new Dictionary<string, object?>
+            {
+                ["session_id"] = _options.SessionId,
+                ["event_family"] = family,
+                ["worker_sequence"] = workerSequence,
+                ["server_handle"] = serverHandle,
+                ["item_handle"] = itemHandle,
+                ["max_message_bytes"] = _options.MaxMessageBytes,
+                // Sizes only — the event value never reaches the log.
+                ["reason"] = exception.Message,
+            });
+
+        string diagnosticMessage =
+            $"{family} event for server handle {serverHandle}, item handle {itemHandle} "
+            + $"(worker sequence {workerSequence}) exceeds the negotiated frame maximum of "
+            + $"{_options.MaxMessageBytes} bytes and cannot be delivered; raise "
+            + "MxGateway:Worker:MaxMessageBytes for this workload.";
+
+        _state = WorkerState.Faulted;
+        await TryWriteFaultAsync(
+            new WorkerFault
+            {
+                Category = WorkerFaultCategory.ProtocolViolation,
+                CommandMethod = "EventDrain",
+                ExceptionType = exception.GetType().FullName ?? string.Empty,
+                DiagnosticMessage = diagnosticMessage,
+                ProtocolStatus = new ProtocolStatus
+                {
+                    Code = ProtocolStatusCode.ProtocolViolation,
+                    Message = diagnosticMessage,
+                },
+            },
+            cancellationToken).ConfigureAwait(false);
+
+        throw new InvalidOperationException(diagnosticMessage, exception);
+    }
+
     private async Task<bool> DispatchGatewayEnvelopeAsync(
         WorkerEnvelope envelope,
         CancellationToken cancellationToken)
@@ -478,7 +548,8 @@ public sealed class WorkerPipeSession
             _ => CreateControlOkReply(correlationId, command.Kind),
         };
 
-        await WriteControlReplyAsync(reply, cancellationToken).ConfigureAwait(false);
+        await WriteControlReplyWithSizeBackstopAsync(reply, correlationId, command.Kind, cancellationToken)
+            .ConfigureAwait(false);
         return true;
     }
 
@@ -495,6 +566,105 @@ public sealed class WorkerPipeSession
             cancellationToken);
     }
 
+    /// <summary>
+    ///     Writes a control reply, answering the correlation with a small error reply instead of
+    ///     unwinding the session if the reply does not fit the negotiated frame maximum. Reply
+    ///     builders already size their payloads (see <see cref="CreateDrainEventsReply"/>), so this
+    ///     is a backstop against a future command or a sizing bug — but without it a single
+    ///     oversized diagnostic reply is session-fatal, which no diagnostics command may be.
+    /// </summary>
+    private async Task WriteControlReplyWithSizeBackstopAsync(
+        MxCommandReply reply,
+        string correlationId,
+        MxCommandKind kind,
+        CancellationToken cancellationToken)
+    {
+        try
+        {
+            await WriteControlReplyAsync(reply, cancellationToken).ConfigureAwait(false);
+        }
+        catch (WorkerFrameProtocolException exception)
+            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(
+        string correlationId,
+        MxCommandKind kind,
+        WorkerFrameProtocolException exception)
+    {
+        _logger?.Error(
+            "WorkerControlReplyTooLarge",
+            new Dictionary<string, object?>
+            {
+                ["correlation_id"] = correlationId,
+                ["command_kind"] = kind.ToString(),
+                ["max_message_bytes"] = _options.MaxMessageBytes,
+                // The writer's message carries the rejected payload length; it names sizes only,
+                // never reply content.
+                ["reason"] = exception.Message,
+            });
+    }
+
+    private MxCommandReply CreateReplyTooLargeReply(string correlationId, MxCommandKind kind)
+    {
+        const string message =
+            "Worker reply exceeded the negotiated frame maximum; retry with a smaller request.";
+        return new MxCommandReply
+        {
+            SessionId = _options.SessionId,
+            CorrelationId = correlationId,
+            Kind = kind,
+            Hresult = 0,
+            DiagnosticMessage = message,
+            ProtocolStatus = new ProtocolStatus
+            {
+                Code = ProtocolStatusCode.InvalidRequest,
+                Message = message,
+            },
+        };
+    }
+
     private MxCommandReply CreatePingReply(string correlationId, MxCommand command)
     {
         MxCommandReply reply = CreateControlOkReply(correlationId, command.Kind);
@@ -543,24 +713,77 @@ public sealed class WorkerPipeSession
         if (runtimeSession is not null)
         {
             // Bound the diagnostic drain so max_events = 0 ("as many as available") or an over-large
-            // request cannot pack the whole queue into one session-killing reply frame.
+            // request cannot pack the whole queue into one session-killing reply frame. The count cap
+            // alone is not enough: byte-heavy events overshoot the negotiated frame maximum long
+            // before the count ceiling, so the drain is also byte-budgeted and sizes the reply while
+            // draining — an event that does not fit is left queued rather than dequeued and lost.
             uint requested = command.DrainEvents?.MaxEvents ?? 0;
-            uint maxEvents = requested == 0 || requested > MaxDrainEventsPerReply
-                ? MaxDrainEventsPerReply
+            uint maxEvents = requested == 0 || requested > GatewayContractInfo.MaxDrainEventsPerCommand
+                ? GatewayContractInfo.MaxDrainEventsPerCommand
                 : requested;
-            foreach (WorkerEvent workerEvent in runtimeSession.DrainEvents(maxEvents))
+            WorkerEventDrainResult drainResult = runtimeSession.DrainEvents(
+                maxEvents,
+                ResolveDrainReplyByteBudget());
+            foreach (WorkerEvent workerEvent in drainResult.Events)
             {
                 if (workerEvent.Event is not null)
                 {
                     drainReply.Events.Add(workerEvent.Event);
                 }
             }
+
+            if (drainResult.TruncatedBySize)
+            {
+                // DrainEventsReply has no truncation field, and adding one would regenerate every
+                // language client for a diagnostic nicety. The reply's existing DiagnosticMessage
+                // carries the same information at zero contract cost; the caller contract is to
+                // repeat DrainEvents until it comes back empty.
+                reply.DiagnosticMessage = CreateDrainTruncationMessage(
+                    drainReply.Events.Count,
+                    drainResult);
+            }
         }
 
         reply.DrainEvents = drainReply;
         return reply;
     }
 
+    /// <summary>
+    ///     Byte budget for the events packed into one DrainEvents reply: the negotiated frame
+    ///     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()
+    {
+        return Math.Max(
+            _options.MaxMessageBytes - DrainReplyFrameHeadroomBytes,
+            _options.MaxMessageBytes / 2);
+    }
+
+    private static string CreateDrainTruncationMessage(
+        int returnedCount,
+        WorkerEventDrainResult drainResult)
+    {
+        string message =
+            $"{returnedCount} events returned, {drainResult.RemainingCount} remain; "
+            + "repeat DrainEvents for the rest.";
+        if (drainResult.OversizedHeadSequence != 0)
+        {
+            message +=
+                $" The next event (worker sequence {drainResult.OversizedHeadSequence}) alone exceeds "
+                + "the negotiated frame maximum and cannot be drained; raise MxGateway:Worker:MaxMessageBytes.";
+        }
+
+        return message;
+    }
+
     private MxCommandReply CreateControlOkReply(string correlationId, MxCommandKind kind)
     {
         return new MxCommandReply
@@ -627,15 +850,30 @@ public sealed class WorkerPipeSession
                 return;
             }
 
-            await _writer
-                .WriteAsync(
-                    CreateEnvelope(new WorkerCommandReply
-                    {
-                        Reply = reply,
-                        CompletedTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
-                    }),
-                    cancellationToken)
-                .ConfigureAwait(false);
+            try
+            {
+                await _writer
+                    .WriteAsync(
+                        CreateEnvelope(new WorkerCommandReply
+                        {
+                            Reply = reply,
+                            CompletedTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+                        }),
+                        cancellationToken)
+                    .ConfigureAwait(false);
+            }
+            catch (WorkerFrameProtocolException sizeException)
+                when (sizeException.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge)
+            {
+                // 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. 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 WriteReplyTooLargeFallbackAsync(envelope.CorrelationId, command.Kind, cancellationToken)
+                    .ConfigureAwait(false);
+            }
         }
         catch (Exception exception) when (exception is not OperationCanceledException)
         {

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

@@ -44,6 +44,19 @@ public interface IWorkerRuntimeSession : IDisposable
     /// <returns>List of drained events.</returns>
     IReadOnlyList<WorkerEvent> DrainEvents(uint maxEvents);
 
+    /// <summary>
+    /// Drains pending events bounded by both a count cap and a byte budget, so a caller building a
+    /// single reply frame never removes an event it cannot ship.
+    /// </summary>
+    /// <remarks>
+    ///     Declared as a second method rather than a default interface method: the worker targets
+    ///     .NET Framework 4.8, which has no runtime support for default interface members.
+    /// </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>
+    /// <returns>The drained events and the truncation facts describing what stayed queued.</returns>
+    WorkerEventDrainResult DrainEvents(uint maxEvents, int maxTotalBytes);
+
     /// <summary>
     /// Drains a pending fault from the queue, if any.
     /// </summary>

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

@@ -26,6 +26,15 @@ public sealed class MxAccessEventQueue
     /// </summary>
     public const int DefaultCapacity = 10000;
 
+    // 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;
     private readonly Queue<WorkerEvent> events;
     private readonly object syncRoot = new();
@@ -209,6 +218,64 @@ public sealed class MxAccessEventQueue
         }
     }
 
+    /// <summary>
+    /// Drains from the head while both the count cap and a byte budget allow it, so the caller can
+    /// build a reply frame that is guaranteed to fit the negotiated frame maximum.
+    /// </summary>
+    /// <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> — 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>
+    /// <returns>The drained events plus the truncation facts the caller reports to the gateway.</returns>
+    public WorkerEventDrainResult Drain(uint maxEvents, int maxTotalBytes)
+    {
+        lock (syncRoot)
+        {
+            int countLimit = maxEvents == 0
+                ? int.MaxValue
+                : checked((int)Math.Min(maxEvents, int.MaxValue));
+            List<WorkerEvent> drained = new();
+            int remainingBudget = maxTotalBytes;
+            bool truncatedBySize = false;
+            ulong oversizedHeadSequence = 0;
+
+            while (drained.Count < countLimit && events.Count > 0)
+            {
+                WorkerEvent head = events.Peek();
+                int cost = head.CalculateSize() + RepeatedFieldOverheadBytes;
+                if (cost > remainingBudget)
+                {
+                    truncatedBySize = true;
+                    if (cost > maxTotalBytes)
+                    {
+                        // The head alone cannot fit this budget, so repeating the call will not
+                        // move it either. Report its sequence instead of silently stalling; the
+                        // events that did fit are still returned.
+                        oversizedHeadSequence = head.Event?.WorkerSequence ?? 0;
+                    }
+
+                    break;
+                }
+
+                remainingBudget -= cost;
+                drained.Add(events.Dequeue());
+            }
+
+            return new WorkerEventDrainResult(
+                drained,
+                truncatedBySize,
+                events.Count,
+                oversizedHeadSequence);
+        }
+    }
+
     /// <summary>
     /// Records a fault if one has not already been recorded.
     /// </summary>

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

@@ -392,6 +392,12 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
         return eventQueue.Drain(maxEvents);
     }
 
+    /// <inheritdoc />
+    public WorkerEventDrainResult DrainEvents(uint maxEvents, int maxTotalBytes)
+    {
+        return eventQueue.Drain(maxEvents, maxTotalBytes);
+    }
+
     /// <inheritdoc />
     public WorkerFault? DrainFault()
     {

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

@@ -0,0 +1,56 @@
+using System.Collections.Generic;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+
+namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
+
+/// <summary>
+/// Outcome of a byte-budgeted drain from the MXAccess outbound event queue.
+/// </summary>
+/// <remarks>
+///     A count cap alone cannot keep a <c>DrainEvents</c> reply inside the negotiated frame
+///     maximum: byte-heavy events (large string or array <c>MxValue</c>s) overshoot the frame max
+///     long before the count ceiling is reached, and the writer's per-frame rejection then
+///     destroys events that were already removed from the queue. The byte-budgeted drain sizes
+///     the reply while draining, so an event that does not fit is never dequeued (WRK-21), and
+///     this result carries the truncation facts the reply's <c>DiagnosticMessage</c> reports —
+///     no contract change is needed to express them.
+///     Plain constructor and get-only properties: the worker targets .NET Framework 4.8, which
+///     has no init-only members or positional records.
+/// </remarks>
+public sealed class WorkerEventDrainResult
+{
+    /// <summary>Initializes a new instance of the <see cref="WorkerEventDrainResult"/> class.</summary>
+    /// <param name="events">Events removed from the queue, in enqueue order.</param>
+    /// <param name="truncatedBySize">Whether the byte budget, not the count cap, ended the drain.</param>
+    /// <param name="remainingCount">Number of events still queued after the drain.</param>
+    /// <param name="oversizedHeadSequence">
+    ///     Worker sequence of a head event whose own serialized size exceeds the whole budget, so
+    ///     no future call of the same budget can ship it; 0 when there is no such event.
+    /// </param>
+    public WorkerEventDrainResult(
+        IReadOnlyList<WorkerEvent> events,
+        bool truncatedBySize,
+        int remainingCount,
+        ulong oversizedHeadSequence)
+    {
+        Events = events;
+        TruncatedBySize = truncatedBySize;
+        RemainingCount = remainingCount;
+        OversizedHeadSequence = oversizedHeadSequence;
+    }
+
+    /// <summary>Gets the events removed from the queue, in enqueue order.</summary>
+    public IReadOnlyList<WorkerEvent> Events { get; }
+
+    /// <summary>Gets a value indicating whether the byte budget ended the drain early.</summary>
+    public bool TruncatedBySize { get; }
+
+    /// <summary>Gets the number of events still queued after the drain.</summary>
+    public int RemainingCount { get; }
+
+    /// <summary>
+    /// Gets the worker sequence of the head event that alone exceeds the byte budget, or 0 when
+    /// no single event blocks the drain. Naming it lets an operator find the offending tag.
+    /// </summary>
+    public ulong OversizedHeadSequence { get; }
+}