Merge pull request 'Retain fake worker fault diagnostics in test fixtures' (#155) from fix/fake-worker-fault-diagnostics into main

src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndMultiSubscriberTests.cs

@@ -2,6 +2,7 @@ using Google.Protobuf.WellKnownTypes;
 using Grpc.Core;
 using Microsoft.Extensions.Logging.Abstractions;
 using Microsoft.Extensions.Options;
+using Xunit.Abstractions;
 using ZB.MOM.WW.MxGateway.Contracts;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
 using ZB.MOM.WW.MxGateway.Server.Configuration;
@@ -21,7 +22,7 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway;
 /// the fake worker harness. Covers fan-out to two concurrent subscribers, independent
 /// cancellation isolation, and the per-session subscriber cap.
 /// </summary>
-public sealed class GatewayEndToEndMultiSubscriberTests
+public sealed class GatewayEndToEndMultiSubscriberTests(ITestOutputHelper output)
 {
     private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10);
     private const int ServerHandle = 3001;
@@ -37,7 +38,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
     public async Task StreamEvents_TwoSubscribers_BothReceiveAllEvents()
     {
         MultiEventFakeWorkerProcessLauncher launcher = new(EventCount);
-        await using MultiSubscriberGatewayServiceFixture fixture = new(launcher);
+        await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, diagnosticOutput: output.WriteLine);
 
         OpenSessionReply openReply = await fixture.Service.OpenSession(
             new OpenSessionRequest
@@ -136,7 +137,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
     public async Task StreamEvents_OneSubscriberCancels_OtherContinuesReceivingEvents()
     {
         GatedEventFakeWorkerProcessLauncher launcher = new();
-        await using MultiSubscriberGatewayServiceFixture fixture = new(launcher);
+        await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, diagnosticOutput: output.WriteLine);
 
         OpenSessionReply openReply = await fixture.Service.OpenSession(
             new OpenSessionRequest
@@ -254,7 +255,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
         // (EventSubscriberLimitReached) is only reachable when multi-subscriber mode is on.
         // Without it the service hits EventSubscriberAlreadyActive instead — also
         // ResourceExhausted, but for the wrong reason — masking a misconfiguration.
-        await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, maxEventSubscribersPerSession: cap);
+        await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, maxEventSubscribersPerSession: cap, diagnosticOutput: output.WriteLine);
 
         OpenSessionReply openReply = await fixture.Service.OpenSession(
             new OpenSessionRequest
@@ -413,20 +414,25 @@ public sealed class GatewayEndToEndMultiSubscriberTests
     {
         private readonly GatewayMetrics _metrics = new();
         private readonly SessionRegistry _registry = new();
+        private readonly FakeWorkerDiagnostics _diagnostics;
 
         /// <summary>Initializes a new instance of the <see cref="MultiSubscriberGatewayServiceFixture"/> class.</summary>
         /// <param name="launcher">Fake worker process launcher backing the session manager.</param>
+        /// <param name="diagnosticOutput">Safe metadata output sink.</param>
         /// <param name="maxEventSubscribersPerSession">Maximum concurrent event subscribers allowed per session.</param>
         public MultiSubscriberGatewayServiceFixture(
             IWorkerProcessLauncher launcher,
-            int maxEventSubscribersPerSession = 8)
+            int maxEventSubscribersPerSession = 8,
+            Action<string>? diagnosticOutput = null)
         {
+            _diagnostics = ((IDiagnosticFakeWorker)launcher).Diagnostics;
+            _diagnostics.SetOutput(diagnosticOutput);
             IOptions<GatewayOptions> options = Options.Create(CreateOptions(maxEventSubscribersPerSession));
             SessionWorkerClientFactory workerClientFactory = new(
                 launcher,
                 options,
                 _metrics,
-                NullLoggerFactory.Instance);
+                _diagnostics);
             SessionManager sessionManager = new(
                 _registry,
                 workerClientFactory,
@@ -499,13 +505,20 @@ public sealed class GatewayEndToEndMultiSubscriberTests
         /// <summary>Disposes every session in the registry and releases the fixture's metrics.</summary>
         /// <returns>A task that represents the asynchronous operation.</returns>
         public async ValueTask DisposeAsync()
+        {
+            // Session disposal and worker observation share the existing test budget.
+            // The observer consumes a late cleanup fault without replacing the primary failure.
+            _diagnostics.Track(DisposeSessionsAsync(), "teardown");
+            await _diagnostics.DrainAsync(TestTimeout).ConfigureAwait(false);
+            _metrics.Dispose();
+        }
+
+        private async Task DisposeSessionsAsync()
         {
             foreach (GatewaySession session in _registry.Snapshot())
             {
-                await session.DisposeAsync();
+                await session.DisposeAsync().ConfigureAwait(false);
             }
-
-            _metrics.Dispose();
         }
 
         private static GatewayOptions CreateOptions(int maxEventSubscribersPerSession) =>
@@ -540,7 +553,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
     /// command, then waits for shutdown. Events carry an indexed string value so the test can
     /// verify fan-out order across two subscribers.
     /// </summary>
-    private sealed class MultiEventFakeWorkerProcessLauncher(int eventCount) : IWorkerProcessLauncher
+    private sealed class MultiEventFakeWorkerProcessLauncher(int eventCount) : IWorkerProcessLauncher, IDiagnosticFakeWorker
     {
         public const int ProcessId = 7710;
 
@@ -549,12 +562,16 @@ public sealed class GatewayEndToEndMultiSubscriberTests
         /// <summary>Gets the task representing the fake worker's running background loop.</summary>
         public Task WorkerTask { get; private set; } = Task.CompletedTask;
 
+        /// <inheritdoc />
+        public FakeWorkerDiagnostics Diagnostics { get; } = new();
+
         /// <inheritdoc />
         public Task<WorkerProcessHandle> LaunchAsync(
             WorkerProcessLaunchRequest request,
             CancellationToken cancellationToken = default)
         {
             WorkerTask = RunWorkerAsync(request, cancellationToken);
+            Diagnostics.Track(WorkerTask);
 
             return Task.FromResult(new WorkerProcessHandle(
                 _process,
@@ -578,6 +595,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
             {
                 WorkerEnvelope envelope = await harness.ReadGatewayEnvelopeAsync(cancellationToken).ConfigureAwait(false);
 
+                Diagnostics.Received(envelope);
                 if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown)
                 {
                     await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
@@ -630,7 +648,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
     /// deterministic. Call <see cref="StopEmitting"/> before closing the session so the
     /// worker loop exits cleanly and can process the shutdown envelope.
     /// </summary>
-    private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher
+    private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher, IDiagnosticFakeWorker
     {
         public const int ProcessId = 7720;
 
@@ -643,6 +661,9 @@ public sealed class GatewayEndToEndMultiSubscriberTests
         /// <summary>Gets the task representing the fake worker's running background loop.</summary>
         public Task WorkerTask { get; private set; } = Task.CompletedTask;
 
+        /// <inheritdoc />
+        public FakeWorkerDiagnostics Diagnostics { get; } = new();
+
         /// <summary>Releases the gate so the worker emits one event.</summary>
         public void AllowNextEvent() => _emitGate.Release();
 
@@ -662,6 +683,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
             CancellationToken cancellationToken = default)
         {
             WorkerTask = RunWorkerAsync(request, cancellationToken);
+            Diagnostics.Track(WorkerTask);
 
             return Task.FromResult(new WorkerProcessHandle(
                 _process,
@@ -717,6 +739,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests
                     continue;
                 }
 
+                Diagnostics.Received(envelope);
                 if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown)
                 {
                     await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false);

src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs

@@ -1,8 +1,10 @@
+using System.Collections.Concurrent;
 using Google.Protobuf.WellKnownTypes;
 using Grpc.Core;
 using Microsoft.Extensions.Logging.Abstractions;
 using Microsoft.Extensions.Options;
 using Microsoft.Extensions.Time.Testing;
+using Xunit.Abstractions;
 using ZB.MOM.WW.MxGateway.Contracts;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
 using ZB.MOM.WW.MxGateway.Server.Configuration;
@@ -36,7 +38,7 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway;
 ///     per session and survives detach, so events emitted while no subscriber is attached are
 ///     retained for the reconnecting stream.
 /// </remarks>
-public sealed class GatewayEndToEndReconnectReplayTests
+public sealed class GatewayEndToEndReconnectReplayTests(ITestOutputHelper output)
 {
     private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10);
     private const int ServerHandle = 3001;
@@ -58,7 +60,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
         GatedEventFakeWorkerProcessLauncher launcher = new();
         // Capacity 16 retains every event emitted here (7 total), so nothing is evicted and a
         // resume from a middle cursor never produces a gap.
-        await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: 16);
+        await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: 16, diagnosticOutput: output.WriteLine);
 
         string sessionId = await OpenSessionAsync(fixture, "reconnect-no-gap");
 
@@ -157,7 +159,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
         GatedEventFakeWorkerProcessLauncher launcher = new();
         // Small capacity forces eviction: with 6 events emitted and a 3-slot ring, the first 3
         // are evicted and only the newest 3 remain replayable.
-        await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: capacity);
+        await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: capacity, diagnosticOutput: output.WriteLine);
 
         string sessionId = await OpenSessionAsync(fixture, "reconnect-gap");
 
@@ -256,7 +258,8 @@ public sealed class GatewayEndToEndReconnectReplayTests
             launcher,
             replayBufferCapacity: 16,
             replayRetentionSeconds: retentionSeconds,
-            timeProvider: clock);
+            timeProvider: clock,
+            diagnosticOutput: output.WriteLine);
 
         string sessionId = await OpenSessionAsync(fixture, "reconnect-aged-out");
 
@@ -340,6 +343,34 @@ public sealed class GatewayEndToEndReconnectReplayTests
             $"Live event {delivered.WorkerSequence} must be newer than the highest previously seen {highestSeen}.");
     }
 
+    /// <summary>Setup failures preserve gateway EOF and report the original fake worker fault safely.</summary>
+    /// <param name="faultCommand">The setup command where the fake worker fails.</param>
+    /// <returns>The asynchronous test operation.</returns>
+    [Theory]
+    [InlineData(MxCommandKind.Register)]
+    [InlineData(MxCommandKind.AddItem)]
+    public async Task Setup_WhenFakeWorkerFaults_ReportsFaultWithoutMaskingRpcFailure(MxCommandKind faultCommand)
+    {
+        ConcurrentQueue<string> output = new();
+        GatedEventFakeWorkerProcessLauncher launcher = new(faultCommand);
+        Exception? failure = await Record.ExceptionAsync(async () =>
+        {
+            await using ReconnectReplayGatewayServiceFixture fixture = new(
+                launcher, replayBufferCapacity: 16, diagnosticOutput: output.Enqueue);
+            string sessionId = await OpenSessionAsync(fixture, "injected-fault");
+            await WireUpAdviseAsync(fixture, sessionId);
+        });
+
+        RpcException rpc = Assert.IsType<RpcException>(failure);
+        Assert.Equal(StatusCode.Unavailable, rpc.StatusCode);
+        Assert.Contains("Worker pipe disconnected", rpc.Status.Detail);
+        Assert.Contains(output, line => line.Contains("source=fake-worker")
+            && line.Contains("type=WorkerFrameProtocolException") && line.Contains("code=InvalidEnvelope")
+            && line.Contains($"command={faultCommand}") && line.Contains("phase=command"));
+        Assert.Contains(output, line => line.Contains("source=gateway") && line.Contains("code=EndOfStream"));
+        Assert.DoesNotContain(output, line => line.Contains("secret-item") || line.Contains("Galaxy.Tag.Value"));
+    }
+
     // ---- shared flow helpers ----
 
     private static async Task<string> OpenSessionAsync(
@@ -490,6 +521,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
     {
         private readonly GatewayMetrics _metrics = new();
         private readonly SessionRegistry _registry = new();
+        private readonly FakeWorkerDiagnostics _diagnostics;
 
         /// <summary>Initializes a new instance of the <see cref="ReconnectReplayGatewayServiceFixture"/> class.</summary>
         /// <param name="launcher">Fake worker process launcher backing the session manager.</param>
@@ -499,6 +531,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
         ///     duration of a fast test; the age-eviction test shortens it and drives
         ///     <paramref name="timeProvider"/>.
         /// </param>
+        /// <param name="diagnosticOutput">Safe metadata output sink.</param>
         /// <param name="timeProvider">
         ///     Clock handed to the session manager, which flows through to the session's event
         ///     distributor. Pass a fake to make age-eviction deterministic.
@@ -507,15 +540,18 @@ public sealed class GatewayEndToEndReconnectReplayTests
             IWorkerProcessLauncher launcher,
             int replayBufferCapacity,
             double replayRetentionSeconds = 300,
-            TimeProvider? timeProvider = null)
+            TimeProvider? timeProvider = null,
+            Action<string>? diagnosticOutput = null)
         {
+            _diagnostics = ((IDiagnosticFakeWorker)launcher).Diagnostics;
+            _diagnostics.SetOutput(diagnosticOutput);
             IOptions<GatewayOptions> options = Options.Create(
                 CreateOptions(replayBufferCapacity, replayRetentionSeconds));
             SessionWorkerClientFactory workerClientFactory = new(
                 launcher,
                 options,
                 _metrics,
-                NullLoggerFactory.Instance);
+                _diagnostics);
             SessionManager sessionManager = new(
                 _registry,
                 workerClientFactory,
@@ -585,13 +621,20 @@ public sealed class GatewayEndToEndReconnectReplayTests
         /// <summary>Disposes every session in the registry and releases the fixture's metrics.</summary>
         /// <returns>A task that represents the asynchronous operation.</returns>
         public async ValueTask DisposeAsync()
+        {
+            // Session disposal and worker observation share the existing test budget.
+            // The observer consumes a late cleanup fault without replacing the primary failure.
+            _diagnostics.Track(DisposeSessionsAsync(), "teardown");
+            await _diagnostics.DrainAsync(TestTimeout).ConfigureAwait(false);
+            _metrics.Dispose();
+        }
+
+        private async Task DisposeSessionsAsync()
         {
             foreach (GatewaySession session in _registry.Snapshot())
             {
                 await session.DisposeAsync().ConfigureAwait(false);
             }
-
-            _metrics.Dispose();
         }
 
         private static GatewayOptions CreateOptions(int replayBufferCapacity, double replayRetentionSeconds) =>
@@ -642,7 +685,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
     ///     <see cref="StopEmitting"/> before closing the session so the loop exits cleanly and can
     ///     process the shutdown envelope.
     /// </summary>
-    private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher
+    private sealed class GatedEventFakeWorkerProcessLauncher(MxCommandKind? faultCommand = null) : IWorkerProcessLauncher, IDiagnosticFakeWorker
     {
         public const int ProcessId = 7730;
 
@@ -655,6 +698,9 @@ public sealed class GatewayEndToEndReconnectReplayTests
         /// <summary>Gets the task representing the fake worker's running background loop.</summary>
         public Task WorkerTask { get; private set; } = Task.CompletedTask;
 
+        /// <inheritdoc />
+        public FakeWorkerDiagnostics Diagnostics { get; } = new();
+
         /// <summary>Releases the gate so the worker emits one event.</summary>
         public void AllowNextEvent() => _emitGate.Release();
 
@@ -674,6 +720,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
             CancellationToken cancellationToken = default)
         {
             WorkerTask = RunWorkerAsync(request, cancellationToken);
+            Diagnostics.Track(WorkerTask);
 
             return Task.FromResult(new WorkerProcessHandle(
                 _process,
@@ -729,6 +776,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
                     continue;
                 }
 
+                Diagnostics.Received(envelope);
                 if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown)
                 {
                     await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
@@ -742,6 +790,11 @@ public sealed class GatewayEndToEndReconnectReplayTests
                 }
 
                 MxCommand command = envelope.WorkerCommand.Command;
+                if (command.Kind == faultCommand)
+                {
+                    throw new WorkerFrameProtocolException(WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item");
+                }
+
                 await harness.ReplyToCommandAsync(
                     envelope,
                     configureReply: reply => ConfigureCommandReply(reply, command.Kind),

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/FakeWorkerDiagnosticsTests.cs

@@ -0,0 +1,88 @@
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.MxGateway.Server.Workers;
+using ZB.MOM.WW.MxGateway.Tests.TestSupport;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Workers;
+
+/// <summary>Protects bounded teardown, fault observation, and safe diagnostic output.</summary>
+public sealed class FakeWorkerDiagnosticsTests
+{
+    /// <summary>Repeated gateway logs remain bounded and never format sensitive state or messages.</summary>
+    [Fact]
+    public void Log_WhenFaultsRepeat_ReportsOnlyBoundedMetadata()
+    {
+        using FakeWorkerDiagnostics diagnostics = new();
+        List<string> output = [];
+        diagnostics.SetOutput(output.Add);
+        for (int i = 0; i < 32; i++)
+        {
+            diagnostics.Log(LogLevel.Warning, new EventId(1), "secret-command",
+                new WorkerFrameProtocolException(WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item"),
+                (state, exception) => $"{state} {exception}");
+        }
+
+        Assert.NotEmpty(output);
+        Assert.InRange(output.Count, 1, 16);
+        Assert.All(output, line =>
+        {
+            Assert.Contains("type=WorkerFrameProtocolException", line);
+            Assert.Contains("code=InvalidEnvelope", line);
+            Assert.DoesNotContain("secret", line);
+        });
+    }
+
+    /// <summary>A late fault is observed after timeout without calling the completed output sink.</summary>
+    /// <returns>The asynchronous test operation.</returns>
+    [Fact]
+    public async Task DrainAsync_WhenWorkerOutlivesBudget_ObservesLateFaultWithoutLateOutput()
+    {
+        FakeWorkerDiagnostics diagnostics = new();
+        List<string> output = [];
+        diagnostics.SetOutput(output.Add);
+        TaskCompletionSource worker = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        diagnostics.Track(worker.Task);
+        await diagnostics.DrainAsync(TimeSpan.Zero);
+        Assert.Contains(output, line => line.Contains("source=teardown") && line.Contains("type=TimeoutException"));
+        int beforeFault = output.Count;
+        worker.SetException(new WorkerFrameProtocolException(WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item"));
+        await diagnostics.DrainAsync(TimeSpan.FromSeconds(5));
+        Assert.Equal(beforeFault, output.Count);
+    }
+
+    /// <summary>Expected cancellation does not add fault diagnostics.</summary>
+    /// <returns>The asynchronous test operation.</returns>
+    [Fact]
+    public async Task DrainAsync_WhenWorkerIsCancelled_DoesNotReportFault()
+    {
+        FakeWorkerDiagnostics diagnostics = new();
+        List<string> output = [];
+        diagnostics.SetOutput(output.Add);
+        diagnostics.Track(Task.FromCanceled(new CancellationToken(true)));
+        await diagnostics.DrainAsync(TimeSpan.FromSeconds(5));
+        Assert.Empty(output);
+    }
+
+    /// <summary>An unavailable output helper cannot fault the observer or replace the primary failure.</summary>
+    /// <returns>The asynchronous test operation.</returns>
+    [Fact]
+    public async Task DrainAsync_WhenOutputThrows_PreservesPrimaryFailure()
+    {
+        InvalidOperationException primary = new("primary");
+        FakeWorkerDiagnostics diagnostics = new();
+        diagnostics.SetOutput(_ => throw new InvalidOperationException("completed output"));
+        Exception? failure = await Record.ExceptionAsync(async () =>
+        {
+            try
+            {
+                throw primary;
+            }
+            finally
+            {
+                diagnostics.Track(Task.FromException(new WorkerFrameProtocolException(
+                    WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item")));
+                await diagnostics.DrainAsync(TimeSpan.FromSeconds(5));
+            }
+        });
+        Assert.Same(primary, failure);
+    }
+}

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/GatedWorkerEnvelopeReaderTests.cs

@@ -11,8 +11,11 @@ public sealed class GatedWorkerEnvelopeReaderTests
 {
     /// <summary>A gate wakeup must preserve a partially consumed frame and the following frame.</summary>
     /// <returns>The asynchronous test operation.</returns>
-    [Fact]
+    /// <param name="watchGate">Whether the emission gate is watched (false before Advise).</param>
-    public async Task ReadOrGateAsync_WhenGateOpensDuringSegmentedFrame_PreservesBothFrames()
+    [Theory]
+    [InlineData(true)]
+    [InlineData(false)]
+    public async Task ReadOrGateAsync_WhenGateOpensDuringSegmentedFrame_PreservesBothFrames(bool watchGate)
     {
         WorkerFrameProtocolOptions options = new("segmented-session");
         WorkerEnvelope first = new()
@@ -38,15 +41,27 @@ public sealed class GatedWorkerEnvelopeReaderTests
 
         // The length and first payload byte are consumed before the remaining bytes exist.
         await pipe.Writer.WriteAsync(bytes.AsMemory(0, 5));
-        Task<WorkerEnvelope?> pending = selector.ReadOrGateAsync();
+        Task<WorkerEnvelope?> pending = selector.ReadOrGateAsync(watchGate);
         Assert.False(pending.IsCompleted);
         gate.Release();
-        Assert.Null(await pending);
+        if (watchGate)
+        {
+            Assert.Null(await pending.WaitAsync(TimeSpan.FromSeconds(5)));
+        }
+        else
+        {
+            // Explicit deadline probe: pre-Advise framing must remain pending beyond the old
+            // 50ms cancellation budget, even with the emission gate already open.
+            Task oldReadBudget = Task.Delay(TimeSpan.FromMilliseconds(100), lifetime.Token);
+            Assert.Same(oldReadBudget, await Task.WhenAny(pending, oldReadBudget));
+            await oldReadBudget;
+        }
+
         await pipe.Writer.WriteAsync(bytes.AsMemory(5));
         await pipe.Writer.CompleteAsync();
 
-        Assert.Equal(first, await selector.ReadOrGateAsync());
+        Assert.Equal(first, watchGate ? await selector.ReadOrGateAsync() : await pending.WaitAsync(TimeSpan.FromSeconds(5)));
-        Assert.Equal(second, await selector.ReadOrGateAsync());
+        Assert.Equal(second, await selector.ReadOrGateAsync(watchGate));
     }
 
     /// <summary>Disposal or lifetime cancellation drains the retained read and gate wait.</summary>

src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeWorkerDiagnostics.cs

@@ -0,0 +1,135 @@
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Workers;
+
+namespace ZB.MOM.WW.MxGateway.Tests.TestSupport;
+
+/// <summary>Exposes test-owned worker tasks and safe diagnostic context to fixture cleanup.</summary>
+internal interface IDiagnosticFakeWorker
+{
+    /// <summary>Gets the diagnostic collector for this launcher's workers.</summary>
+    FakeWorkerDiagnostics Diagnostics { get; }
+}
+
+/// <summary>Observes fake worker faults without replacing the test's primary exception.</summary>
+internal sealed class FakeWorkerDiagnostics : ILoggerFactory, ILogger
+{
+    private readonly object _sync = new();
+    private readonly List<Task> _observations = [];
+    private Action<string>? _output;
+    private int _remaining = 16;
+    private bool _closed;
+    private string _phase = "startup";
+    private MxCommandKind? _command;
+    private ulong? _sequence;
+
+    /// <summary>Attaches the fixture's output sink before any workers are launched.</summary>
+    /// <param name="output">Destination for safe, bounded metadata.</param>
+    public void SetOutput(Action<string>? output) => _output = output;
+
+    /// <summary>Records only structural context, never envelopes or command arguments.</summary>
+    /// <param name="envelope">Incoming envelope whose safe metadata is retained.</param>
+    public void Received(WorkerEnvelope envelope)
+    {
+        lock (_sync)
+        {
+            _phase = envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerCommand ? "command" : "shutdown";
+            _command = envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerCommand
+                ? envelope.WorkerCommand.Command.Kind : null;
+            _sequence = envelope.Sequence;
+        }
+    }
+
+    /// <summary>Retains an observation that consumes faults even if bounded teardown times out.</summary>
+    /// <param name="worker">Background task launched by the fake worker.</param>
+    /// <param name="source">Whether the task belongs to a fake worker or fixture teardown.</param>
+    public void Track(Task worker, string source = "fake-worker")
+    {
+        Task observation = worker.ContinueWith(completed =>
+        {
+            // Reading Exception observes every fault; retain no unobserved continuation fault.
+            if (completed.Exception is { } faults)
+            {
+                foreach (Exception fault in faults.Flatten().InnerExceptions)
+                {
+                    Report(source, fault);
+                }
+            }
+        }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
+        _observations.Add(observation);
+    }
+
+    /// <summary>Drains workers under the fixture's existing time limit and closes output.</summary>
+    /// <param name="timeout">Existing test timeout; never extended for diagnostics.</param>
+    /// <returns>The bounded cleanup operation.</returns>
+    public async Task DrainAsync(TimeSpan timeout)
+    {
+        try
+        {
+            await Task.WhenAll(_observations).WaitAsync(timeout).ConfigureAwait(false);
+        }
+        catch (TimeoutException exception)
+        {
+            Report("teardown", exception);
+        }
+        finally
+        {
+            Dispose();
+        }
+    }
+
+    private void Report(string source, Exception exception)
+    {
+        lock (_sync)
+        {
+            if (_closed || _remaining-- <= 0)
+            {
+                return;
+            }
+
+            string code = exception is WorkerFrameProtocolException protocol ? protocol.ErrorCode.ToString() : "none";
+            string line = $"source={source} type={exception.GetType().Name} code={code} phase={_phase} command={_command} sequence={_sequence}";
+            try
+            {
+                _output?.Invoke(line);
+            }
+            catch (Exception)
+            {
+                // A completed output helper must never mask the primary test failure.
+            }
+        }
+    }
+
+    /// <inheritdoc />
+    public ILogger CreateLogger(string categoryName) => this;
+
+    /// <inheritdoc />
+    public void AddProvider(ILoggerProvider provider) => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    public void Dispose()
+    {
+        lock (_sync)
+        {
+            _closed = true;
+            _output = null;
+        }
+    }
+
+    /// <inheritdoc />
+    public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+
+    /// <inheritdoc />
+    public bool IsEnabled(LogLevel logLevel) => logLevel >= LogLevel.Warning;
+
+    /// <inheritdoc />
+    public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+        Func<TState, Exception?, string> formatter)
+    {
+        // Never invoke formatter: state and exception messages may contain item names/secrets.
+        if (IsEnabled(logLevel) && exception is not null)
+        {
+            Report("gateway", exception);
+        }
+    }
+}