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,
@@ -685,19 +707,19 @@ public sealed class GatewayEndToEndMultiSubscriberTests
             int advisedItemHandle = 0;
             int emittedCount = 0;
 
-            // Read envelopes one at a time. Between each envelope, if we have a
-            // subscription and the gate is ready, emit an event before reading the
-            // next envelope. When _stopEmitting is set, drain the gate and read
-            // remaining envelopes (including shutdown) without emitting.
+            await using GatedWorkerEnvelopeReader incoming = new(
+                harness.ReadGatewayEnvelopeAsync, _emitGate, cancellationToken);
             while (!cancellationToken.IsCancellationRequested)
             {
-                // While subscribed and not stopped, try to emit gated events using
-                // a short-timeout peek at the gate — yield to incoming envelopes to
-                // avoid starving shutdown processing.
-                while (advisedServerHandle != 0
-                    && !_stopEmitting
-                    && await _emitGate.WaitAsync(millisecondsTimeout: 0).ConfigureAwait(false))
+                WorkerEnvelope? envelope = await incoming.ReadOrGateAsync(
+                    advisedServerHandle != 0 && !_stopEmitting).ConfigureAwait(false);
+                if (envelope is null)
                 {
+                    if (_stopEmitting)
+                    {
+                        continue;
+                    }
+
                     int index = ++emittedCount;
                     await harness.EmitEventAsync(
                         MxEventFamily.OnDataChange,
@@ -714,23 +736,10 @@ public sealed class GatewayEndToEndMultiSubscriberTests
                             };
                             mxEvent.OnDataChange = new OnDataChangeEvent();
                         }).ConfigureAwait(false);
-                }
-
-                // Use a short timeout so the emit loop above is re-evaluated
-                // periodically — but long enough not to spam.
-                WorkerEnvelope? envelope = null;
-                try
-                {
-                    using CancellationTokenSource readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-                    readCts.CancelAfter(TimeSpan.FromMilliseconds(50));
-                    envelope = await harness.ReadGatewayEnvelopeAsync(readCts.Token).ConfigureAwait(false);
-                }
-                catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
-                {
-                    // Timed out waiting for an envelope — loop back to check gate / emit.
                     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,
@@ -697,14 +744,19 @@ public sealed class GatewayEndToEndReconnectReplayTests
             int advisedItemHandle = 0;
             int emittedCount = 0;
 
+            await using GatedWorkerEnvelopeReader incoming = new(
+                harness.ReadGatewayEnvelopeAsync, _emitGate, cancellationToken);
             while (!cancellationToken.IsCancellationRequested)
             {
-                // While subscribed and not stopped, emit gated events using a non-blocking peek
-                // at the gate so incoming envelopes (including shutdown) are never starved.
-                while (advisedServerHandle != 0
-                    && !_stopEmitting
-                    && await _emitGate.WaitAsync(millisecondsTimeout: 0).ConfigureAwait(false))
+                WorkerEnvelope? envelope = await incoming.ReadOrGateAsync(
+                    advisedServerHandle != 0 && !_stopEmitting).ConfigureAwait(false);
+                if (envelope is null)
                 {
+                    if (_stopEmitting)
+                    {
+                        continue;
+                    }
+
                     int index = ++emittedCount;
                     await harness.EmitEventAsync(
                         MxEventFamily.OnDataChange,
@@ -721,21 +773,10 @@ public sealed class GatewayEndToEndReconnectReplayTests
                             };
                             mxEvent.OnDataChange = new OnDataChangeEvent();
                         }).ConfigureAwait(false);
-                }
-
-                WorkerEnvelope? envelope;
-                try
-                {
-                    using CancellationTokenSource readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-                    readCts.CancelAfter(TimeSpan.FromMilliseconds(50));
-                    envelope = await harness.ReadGatewayEnvelopeAsync(readCts.Token).ConfigureAwait(false);
-                }
-                catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
-                {
-                    // Timed out waiting for an envelope — loop back to check the gate / emit.
                     continue;
                 }
 
+                Diagnostics.Received(envelope);
                 if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown)
                 {
                     await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
@@ -749,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/Sessions/SessionManagerTests.cs

@@ -478,24 +478,42 @@ public sealed class SessionManagerTests
     [Fact]
     public async Task InvokeAsync_WhenWorkerHandshakingThenReadyWithinTimeout_Succeeds()
     {
+        FakeTimeProvider clock = new(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero));
         FakeWorkerClient workerClient = new() { State = WorkerClientState.Handshaking };
         SessionManager manager = CreateManager(
             new FakeSessionWorkerClientFactory(workerClient),
-            options: CreateOptions(workerReadyWaitTimeoutMs: 500));
+            options: CreateOptions(workerReadyWaitTimeoutMs: 500),
+            timeProvider: clock);
         GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
         Assert.Equal(SessionState.Ready, session.State);
 
-        // Flip the worker to Ready shortly after the invoke starts waiting.
-        _ = Task.Run(async () =>
+        TaskCompletionSource handshakingObserved = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        TaskCompletionSource allowReady = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        workerClient.OnStateRead = state =>
         {
-            await Task.Delay(50, CancellationToken.None);
-            workerClient.State = WorkerClientState.Ready;
-        });
-
-        WorkerCommandReply reply = await manager.InvokeAsync(
+            if (state == WorkerClientState.Handshaking)
+            {
+                handshakingObserved.TrySetResult();
+            }
+        };
+        Task transition = TransitionAsync();
+        Task<WorkerCommandReply> invoke = manager.InvokeAsync(
             session.SessionId,
             CreateCommand(MxCommandKind.Ping),
             CancellationToken.None);
+        await handshakingObserved.Task.WaitAsync(TimeSpan.FromSeconds(5));
+        Assert.False(invoke.IsCompleted);
+        Assert.Equal(0, workerClient.InvokeCount);
+        allowReady.SetResult();
+        await transition.WaitAsync(TimeSpan.FromSeconds(5));
+        clock.Advance(TimeSpan.FromMilliseconds(50));
+        WorkerCommandReply reply = await invoke.WaitAsync(TimeSpan.FromSeconds(5));
+
+        async Task TransitionAsync()
+        {
+            await allowReady.Task;
+            workerClient.State = WorkerClientState.Ready;
+        }
 
         Assert.NotNull(reply);
         Assert.Equal(1, workerClient.InvokeCount);
@@ -1406,14 +1424,28 @@ public sealed class SessionManagerTests
 
     private sealed class FakeWorkerClient : IWorkerClient
     {
+        private WorkerClientState _state = WorkerClientState.Ready;
+
         /// <inheritdoc />
         public string SessionId { get; init; } = "session-1";
 
         /// <inheritdoc />
         public int? ProcessId { get; init; } = 1234;
 
+        /// <summary>Gets or sets the optional state-observation gate.</summary>
+        public Action<WorkerClientState>? OnStateRead { get; set; }
+
         /// <inheritdoc />
-        public WorkerClientState State { get; set; } = WorkerClientState.Ready;
+        public WorkerClientState State
+        {
+            get
+            {
+                OnStateRead?.Invoke(_state);
+                return _state;
+            }
+
+            set => _state = value;
+        }
 
         /// <inheritdoc />
         public DateTimeOffset LastHeartbeatAt { get; init; } = DateTimeOffset.UtcNow;

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

@@ -0,0 +1,98 @@
+using System.IO.Pipelines;
+using ZB.MOM.WW.MxGateway.Contracts;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Workers;
+using ZB.MOM.WW.MxGateway.Tests.TestSupport;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Workers;
+
+/// <summary>Protects fake worker framing while the emission gate wakes its loop.</summary>
+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>
+    /// <param name="watchGate">Whether the emission gate is watched (false before Advise).</param>
+    [Theory]
+    [InlineData(true)]
+    [InlineData(false)]
+    public async Task ReadOrGateAsync_WhenGateOpensDuringSegmentedFrame_PreservesBothFrames(bool watchGate)
+    {
+        WorkerFrameProtocolOptions options = new("segmented-session");
+        WorkerEnvelope first = new()
+        {
+            SessionId = "segmented-session",
+            ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
+            Sequence = 1,
+            WorkerShutdown = new WorkerShutdown(),
+        };
+        WorkerEnvelope second = first.Clone();
+        second.Sequence = 2;
+        await using MemoryStream encoded = new();
+        WorkerFrameWriter writer = new(encoded, options);
+        await writer.WriteAsync(first);
+        await writer.WriteAsync(second);
+        byte[] bytes = encoded.ToArray();
+        Pipe pipe = new();
+        await using Stream input = pipe.Reader.AsStream();
+        WorkerFrameReader reader = new(input, options);
+        using SemaphoreSlim gate = new(0);
+        using CancellationTokenSource lifetime = new(TimeSpan.FromSeconds(5));
+        await using GatedWorkerEnvelopeReader selector = new(token => reader.ReadAsync(token).AsTask(), gate, lifetime.Token);
+
+        // 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(watchGate);
+        Assert.False(pending.IsCompleted);
+        gate.Release();
+        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, watchGate ? await selector.ReadOrGateAsync() : await pending.WaitAsync(TimeSpan.FromSeconds(5)));
+        Assert.Equal(second, await selector.ReadOrGateAsync(watchGate));
+    }
+
+    /// <summary>Disposal or lifetime cancellation drains the retained read and gate wait.</summary>
+    /// <param name="cancelLifetime">Whether cancellation precedes disposal.</param>
+    /// <returns>The asynchronous test operation.</returns>
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task DisposeAsync_WithIncompleteFrame_CancelsRetainedRead(bool cancelLifetime)
+    {
+        Pipe pipe = new();
+        await using Stream input = pipe.Reader.AsStream();
+        WorkerFrameReader reader = new(input, new WorkerFrameProtocolOptions("cancel-session"));
+        using SemaphoreSlim gate = new(0);
+        using CancellationTokenSource lifetime = new();
+        Task<WorkerEnvelope>? frameRead = null;
+        GatedWorkerEnvelopeReader selector = new(
+            token => frameRead = reader.ReadAsync(token).AsTask(), gate, lifetime.Token);
+        Task<WorkerEnvelope?> pending = selector.ReadOrGateAsync();
+        gate.Release();
+        Assert.Null(await pending.WaitAsync(TimeSpan.FromSeconds(5)));
+        Assert.NotNull(frameRead);
+        Assert.False(frameRead.IsCompleted);
+        if (cancelLifetime)
+        {
+            await lifetime.CancelAsync();
+        }
+
+        await selector.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5));
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => frameRead);
+        await pipe.Writer.CompleteAsync();
+    }
+
+}

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);
+        }
+    }
+}

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

@@ -0,0 +1,69 @@
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+
+namespace ZB.MOM.WW.MxGateway.Tests.TestSupport;
+
+/// <summary>Coordinates a fake worker's incoming frames and event emission gate.</summary>
+internal sealed class GatedWorkerEnvelopeReader(
+    Func<CancellationToken, Task<WorkerEnvelope>> read,
+    SemaphoreSlim gate,
+    CancellationToken cancellationToken) : IAsyncDisposable
+{
+    private readonly CancellationTokenSource _lifetime = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+    private Task<WorkerEnvelope>? _pendingRead;
+    private Task? _pendingGate;
+
+    /// <summary>Reads an envelope, or returns null when emission should be checked.</summary>
+    /// <param name="watchGate">Whether the worker is currently subscribed and emitting.</param>
+    /// <returns>The next complete envelope, or null for an emission signal.</returns>
+    public async Task<WorkerEnvelope?> ReadOrGateAsync(bool watchGate = true)
+    {
+        // A gate wakeup must never cancel or replace a partially consumed frame.
+        _pendingRead ??= read(_lifetime.Token);
+        if (watchGate)
+        {
+            _pendingGate ??= gate.WaitAsync(_lifetime.Token);
+            await Task.WhenAny(_pendingRead, _pendingGate).ConfigureAwait(false);
+            if (!_pendingRead.IsCompleted)
+            {
+                await _pendingGate.ConfigureAwait(false);
+                _pendingGate = null;
+                return null;
+            }
+        }
+
+        WorkerEnvelope envelope = await _pendingRead.ConfigureAwait(false);
+        _pendingRead = null;
+        return envelope;
+    }
+
+    /// <inheritdoc />
+    public async ValueTask DisposeAsync()
+    {
+        await _lifetime.CancelAsync().ConfigureAwait(false);
+        try
+        {
+            if (_pendingRead is not null)
+            {
+                await _pendingRead.ConfigureAwait(false);
+            }
+        }
+        catch (OperationCanceledException) when (_lifetime.IsCancellationRequested)
+        {
+        }
+        finally
+        {
+            try
+            {
+                if (_pendingGate is not null)
+                {
+                    await _pendingGate.ConfigureAwait(false);
+                }
+            }
+            catch (OperationCanceledException) when (_lifetime.IsCancellationRequested)
+            {
+            }
+
+            _lifetime.Dispose();
+        }
+    }
+}