mxaccessgw/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs

using System.IO.Pipes;
using Google.Protobuf.WellKnownTypes;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Workers;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;

namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Workers;

public sealed class WorkerClientTests
{
    private const string SessionId = "session-worker-client";
    private const string Nonce = "nonce-worker-client";
    private const int WorkerProcessId = 4321;
    private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);

    /// <summary>Verifies that StartAsync enters ready state after receiving worker hello and ready messages.</summary>
    [Fact]
    public async Task StartAsync_WithWorkerHelloAndReady_EntersReadyState()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);

        await CompleteHandshakeAsync(client, pipePair);

        Assert.Equal(WorkerClientState.Ready, client.State);
        Assert.Equal(WorkerProcessId, client.ProcessId);
    }

    /// <summary>Verifies that InvokeAsync completes a pending command when a matching reply arrives.</summary>
    [Fact]
    public async Task InvokeAsync_WithMatchingReply_CompletesPendingCommand()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);

        Task<WorkerCommandReply> invokeTask = client.InvokeAsync(
            CreateCommand(MxCommandKind.Ping),
            TestTimeout,
            CancellationToken.None);

        WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase);
        Assert.False(string.IsNullOrWhiteSpace(commandEnvelope.CorrelationId));

        await pipePair.WorkerWriter.WriteAsync(
            CreateCommandReplyEnvelope(commandEnvelope.CorrelationId, MxCommandKind.Ping));

        WorkerCommandReply reply = await invokeTask.WaitAsync(TestTimeout);

        Assert.Equal(commandEnvelope.CorrelationId, reply.Reply.CorrelationId);
        Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind);
    }

    /// <summary>Verifies that InvokeAsync ignores late replies and keeps the client ready.</summary>
    [Fact]
    public async Task InvokeAsync_WithLateReply_IgnoresLateReplyAndKeepsClientReady()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);

        Task<WorkerCommandReply> timedOutInvokeTask = client.InvokeAsync(
            CreateCommand(MxCommandKind.Ping),
            TimeSpan.FromMilliseconds(50),
            CancellationToken.None);
        WorkerEnvelope timedOutCommand = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);

        WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
            async () => await timedOutInvokeTask);
        Assert.Equal(WorkerClientErrorCode.CommandTimeout, exception.ErrorCode);

        // Send the stale reply for the already-timed-out command, then the second
        // command's reply. The pipe is FIFO, so the read loop processes (and discards)
        // the stale reply before the second reply — no fixed Task.Delay needed.
        await pipePair.WorkerWriter.WriteAsync(
            CreateCommandReplyEnvelope(timedOutCommand.CorrelationId, MxCommandKind.Ping));

        Task<WorkerCommandReply> secondInvokeTask = client.InvokeAsync(
            CreateCommand(MxCommandKind.GetWorkerInfo),
            TestTimeout,
            CancellationToken.None);
        WorkerEnvelope secondCommand = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
        await pipePair.WorkerWriter.WriteAsync(
            CreateCommandReplyEnvelope(secondCommand.CorrelationId, MxCommandKind.GetWorkerInfo));

        WorkerCommandReply reply = await secondInvokeTask.WaitAsync(TestTimeout);

        Assert.Equal(WorkerClientState.Ready, client.State);
        Assert.Equal(MxCommandKind.GetWorkerInfo, reply.Reply.Kind);
    }

    /// <summary>Verifies that ReadEventsAsync yields events in pipe order from the worker.</summary>
    [Fact]
    public async Task ReadEventsAsync_WithWorkerEvents_YieldsEventsInPipeOrder()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);
        using CancellationTokenSource cancellationTokenSource = new(TestTimeout);

        await using IAsyncEnumerator<WorkerEvent> events =
            client.ReadEventsAsync(cancellationTokenSource.Token).GetAsyncEnumerator(cancellationTokenSource.Token);

        await pipePair.WorkerWriter.WriteAsync(
            CreateEventEnvelope(sequence: 11, MxEventFamily.OnDataChange));
        await pipePair.WorkerWriter.WriteAsync(
            CreateEventEnvelope(sequence: 12, MxEventFamily.OperationComplete));

        Assert.True(await events.MoveNextAsync());
        Assert.Equal((ulong)11, events.Current.Event.WorkerSequence);
        Assert.Equal(MxEventFamily.OnDataChange, events.Current.Event.Family);

        Assert.True(await events.MoveNextAsync());
        Assert.Equal((ulong)12, events.Current.Event.WorkerSequence);
        Assert.Equal(MxEventFamily.OperationComplete, events.Current.Event.Family);
    }

    /// <summary>Verifies that the read loop faults the client when the event queue overflows.</summary>
    [Fact]
    public async Task ReadLoop_WhenEventQueueOverflows_FaultsClient()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(
            pipePair,
            new WorkerClientOptions
            {
                EventChannelCapacity = 1,
                EventChannelFullModeTimeout = TimeSpan.FromMilliseconds(50),
                HeartbeatGrace = TimeSpan.FromSeconds(30),
                HeartbeatCheckInterval = TimeSpan.FromSeconds(30),
            });
        await CompleteHandshakeAsync(client, pipePair);

        await pipePair.WorkerWriter.WriteAsync(
            CreateEventEnvelope(sequence: 11, MxEventFamily.OnDataChange));
        await pipePair.WorkerWriter.WriteAsync(
            CreateEventEnvelope(sequence: 12, MxEventFamily.OnDataChange));

        await WaitUntilAsync(
            () => client.State == WorkerClientState.Faulted,
            TestTimeout);

        Assert.Equal(WorkerClientState.Faulted, client.State);
    }

    /// <summary>
    /// Verifies that when the client faults it kills the owned worker process.
    /// The assertion waits on <see cref="FakeWorkerProcess.WaitForExitAsync"/>, which
    /// completes exactly when <c>Kill</c> runs, instead of polling <c>client.State</c>.
    /// Polling state is racy: <see cref="WorkerClient.SetFaulted"/> publishes the
    /// <c>Faulted</c> state before it calls <c>KillOwnedProcess</c>, so a state-based
    /// wait can observe <c>Faulted</c> while <c>KillCount</c> is still 0.
    /// </summary>
    [Fact]
    public async Task ReadLoop_WhenClientFaults_KillsOwnedWorkerProcess()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        FakeWorkerProcess process = new();
        await using WorkerClient client = CreateClient(
            pipePair,
            new WorkerClientOptions
            {
                EventChannelCapacity = 1,
                EventChannelFullModeTimeout = TimeSpan.FromMilliseconds(50),
                HeartbeatGrace = TimeSpan.FromSeconds(30),
                HeartbeatCheckInterval = TimeSpan.FromSeconds(30),
            },
            processHandle: CreateProcessHandle(process));
        await CompleteHandshakeAsync(client, pipePair);

        await pipePair.WorkerWriter.WriteAsync(
            CreateEventEnvelope(sequence: 11, MxEventFamily.OnDataChange));
        await pipePair.WorkerWriter.WriteAsync(
            CreateEventEnvelope(sequence: 12, MxEventFamily.OnDataChange));

        // Deterministic: this completes the instant Kill() runs, with no timing window.
        using CancellationTokenSource exitTimeout = new(TestTimeout);
        await process.WaitForExitAsync(exitTimeout.Token);

        Assert.Equal(WorkerClientState.Faulted, client.State);
        Assert.Equal(1, process.KillCount);
        Assert.True(process.KillEntireProcessTree);
        Assert.True(process.HasExited);
    }

    /// <summary>
    /// Verifies that a worker faulting mid-command — the pipe dropping while an
    /// <see cref="WorkerClient.InvokeAsync"/> is still pending — completes the pending
    /// invoke task with a <see cref="WorkerClientException"/> carrying the
    /// pipe-disconnected error code rather than hanging until the command timeout.
    /// </summary>
    [Fact]
    public async Task InvokeAsync_WhenPipeDisconnectsMidCommand_FailsPendingInvokeWithPipeDisconnected()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);

        Task<WorkerCommandReply> invokeTask = client.InvokeAsync(
            CreateCommand(MxCommandKind.Ping),
            TestTimeout,
            CancellationToken.None);

        // The worker received the command but disconnects before replying.
        WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase);
        await pipePair.DisposeWorkerSideAsync();

        WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
            async () => await invokeTask.WaitAsync(TestTimeout));

        Assert.Equal(WorkerClientErrorCode.PipeDisconnected, exception.ErrorCode);
        await WaitUntilAsync(() => client.State == WorkerClientState.Faulted, TestTimeout);
        Assert.Equal(WorkerClientState.Faulted, client.State);
    }

    /// <summary>
    /// Verifies that a worker emitting a <c>WorkerFault</c> envelope while an
    /// <see cref="WorkerClient.InvokeAsync"/> is pending completes the pending invoke
    /// task with a <see cref="WorkerClientException"/> carrying the worker-faulted
    /// error code.
    /// </summary>
    [Fact]
    public async Task InvokeAsync_WhenWorkerFaultsMidCommand_FailsPendingInvokeWithWorkerFaulted()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);

        Task<WorkerCommandReply> invokeTask = client.InvokeAsync(
            CreateCommand(MxCommandKind.Ping),
            TestTimeout,
            CancellationToken.None);

        WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase);
        await pipePair.WorkerWriter.WriteAsync(CreateWorkerFaultEnvelope("scripted mid-command fault"));

        WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
            async () => await invokeTask.WaitAsync(TestTimeout));

        Assert.Equal(WorkerClientErrorCode.WorkerFaulted, exception.ErrorCode);
        await WaitUntilAsync(() => client.State == WorkerClientState.Faulted, TestTimeout);
        Assert.Equal(WorkerClientState.Faulted, client.State);
    }

    /// <summary>Verifies that pipe disconnect faults the client.</summary>
    [Fact]
    public async Task ReadLoop_WhenPipeDisconnects_FaultsClient()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);

        await pipePair.DisposeWorkerSideAsync();

        await WaitUntilAsync(
            () => client.State == WorkerClientState.Faulted,
            TestTimeout);

        Assert.Equal(WorkerClientState.Faulted, client.State);
    }

    /// <summary>Verifies that the read loop stops the running worker metric when the pipe disconnects.</summary>
    [Fact]
    public async Task ReadLoop_WhenPipeDisconnects_StopsRunningWorkerMetric()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        using GatewayMetrics metrics = new();
        await using WorkerClient client = CreateClient(pipePair, metrics: metrics);
        await CompleteHandshakeAsync(client, pipePair);

        Assert.Equal(1, metrics.GetSnapshot().WorkersRunning);

        await pipePair.DisposeWorkerSideAsync();

        await WaitUntilAsync(
            () => client.State == WorkerClientState.Faulted
                && metrics.GetSnapshot().WorkersRunning == 0,
            TestTimeout);

        GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
        Assert.Equal(0, snapshot.WorkersRunning);
        Assert.Equal(1, snapshot.WorkerExits);
    }

    /// <summary>Verifies that DisposeAsync returns within a bounded timeout when the pipe read is blocked.</summary>
    [Fact]
    public async Task DisposeAsync_WhenPipeReadIsBlocked_ReturnsWithinBoundedTimeout()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);

        DateTimeOffset startedAt = DateTimeOffset.UtcNow;
        await client.DisposeAsync().AsTask().WaitAsync(TestTimeout);
        TimeSpan elapsed = DateTimeOffset.UtcNow - startedAt;

        Assert.True(
            elapsed < TimeSpan.FromSeconds(4),
            $"DisposeAsync took {elapsed.TotalMilliseconds:N0}ms.");
    }

    /// <summary>Verifies that the read loop updates the last heartbeat and worker process when a heartbeat arrives.</summary>
    [Fact]
    public async Task DisposeAsync_WhenOwnedWorkerStillRuns_KillsProcessBeforeDisposing()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        FakeWorkerProcess process = new();
        WorkerClient client = CreateClient(pipePair, processHandle: CreateProcessHandle(process));

        await client.DisposeAsync().AsTask().WaitAsync(TestTimeout);

        Assert.Equal(1, process.KillCount);
        Assert.True(process.KillEntireProcessTree);
        Assert.True(process.Disposed);
    }

    /// <summary>
    /// Verifies that a heartbeat envelope updates the last-heartbeat timestamp and worker
    /// process id. Uses a <see cref="ManualTimeProvider"/> so the timestamp advance is
    /// deterministic instead of relying on a wall-clock <c>Task.Delay</c> exceeding
    /// <see cref="DateTimeOffset.UtcNow"/> resolution.
    /// </summary>
    [Fact]
    public async Task ReadLoop_WhenHeartbeatArrives_UpdatesLastHeartbeatAndWorkerProcess()
    {
        ManualTimeProvider clock = new(DateTimeOffset.Parse("2026-05-18T12:00:00Z", System.Globalization.CultureInfo.InvariantCulture));
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair, timeProvider: clock);
        await CompleteHandshakeAsync(client, pipePair);
        DateTimeOffset previousHeartbeat = client.LastHeartbeatAt;

        clock.Advance(TimeSpan.FromSeconds(1));
        await pipePair.WorkerWriter.WriteAsync(CreateHeartbeatEnvelope(workerProcessId: 9876));

        await WaitUntilAsync(
            () => client.ProcessId == 9876 && client.LastHeartbeatAt > previousHeartbeat,
            TestTimeout);

        Assert.Equal(WorkerClientState.Ready, client.State);
        Assert.Equal(previousHeartbeat + TimeSpan.FromSeconds(1), client.LastHeartbeatAt);
    }

    /// <summary>
    /// Verifies that the heartbeat monitor faults the client when the heartbeat expires.
    /// Uses an injected <see cref="ManualTimeProvider"/> so the grace comparison is deterministic
    /// instead of depending on real wall-clock advance; the monitor's
    /// <see cref="WorkerClientOptions.HeartbeatCheckInterval"/> timer stays on the real clock and
    /// observes the manually-advanced grace on its next tick.
    /// </summary>
    [Fact]
    public async Task HeartbeatMonitor_WhenHeartbeatExpires_FaultsClient()
    {
        ManualTimeProvider clock = new(DateTimeOffset.Parse("2026-05-20T12:00:00Z", System.Globalization.CultureInfo.InvariantCulture));
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(
            pipePair,
            new WorkerClientOptions
            {
                HeartbeatGrace = TimeSpan.FromMilliseconds(80),
                HeartbeatCheckInterval = TimeSpan.FromMilliseconds(20),
                EventChannelCapacity = 8,
            },
            timeProvider: clock);
        await CompleteHandshakeAsync(client, pipePair);

        clock.Advance(TimeSpan.FromSeconds(2));

        await WaitUntilAsync(
            () => client.State == WorkerClientState.Faulted,
            TestTimeout);

        Assert.Equal(WorkerClientState.Faulted, client.State);
    }

    /// <summary>
    ///     Server-031 regression: while a command is in flight on the
    ///     gateway↔worker pipe and the oldest pending command is younger
    ///     than <see cref="WorkerClientOptions.HeartbeatStuckCeiling"/>, the
    ///     heartbeat watchdog must NOT fault on heartbeat-expired alone — the
    ///     gap is more likely caused by pipe-write contention than by a hung
    ///     worker. Mirrors Worker-023 on the worker side.
    /// </summary>
    [Fact]
    public async Task HeartbeatMonitor_WhenCommandInFlightWithinCeiling_DoesNotFaultOnExpiredHeartbeat()
    {
        ManualTimeProvider clock = new(DateTimeOffset.Parse("2026-05-20T13:00:00Z", System.Globalization.CultureInfo.InvariantCulture));
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(
            pipePair,
            new WorkerClientOptions
            {
                HeartbeatGrace = TimeSpan.FromMilliseconds(80),
                HeartbeatCheckInterval = TimeSpan.FromMilliseconds(20),
                EventChannelCapacity = 8,
                HeartbeatStuckCeiling = TimeSpan.FromSeconds(30),
            },
            timeProvider: clock);
        await CompleteHandshakeAsync(client, pipePair);

        // Begin a command that the test never replies to — keeps the
        // PendingCommand alive in `_pendingCommands` for the duration.
        Task<WorkerCommandReply> pendingInvoke = client.InvokeAsync(
            CreateCommand(MxCommandKind.Ping),
            TestTimeout,
            CancellationToken.None);
        WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase);

        // Advance well past HeartbeatGrace but well within HeartbeatStuckCeiling.
        clock.Advance(TimeSpan.FromSeconds(2));

        // Give the heartbeat monitor a few real check-intervals to observe the gap.
        await Task.Delay(TimeSpan.FromMilliseconds(150));

        Assert.Equal(WorkerClientState.Ready, client.State);
        Assert.False(pendingInvoke.IsCompleted);
    }

    /// <summary>
    ///     Server-031 regression: once the oldest pending command exceeds
    ///     <see cref="WorkerClientOptions.HeartbeatStuckCeiling"/>, the
    ///     heartbeat watchdog fires anyway — a truly stuck COM call shouldn't
    ///     keep the watchdog suppressed indefinitely.
    /// </summary>
    [Fact]
    public async Task HeartbeatMonitor_WhenPendingCommandExceedsStuckCeiling_FaultsClient()
    {
        ManualTimeProvider clock = new(DateTimeOffset.Parse("2026-05-20T13:00:00Z", System.Globalization.CultureInfo.InvariantCulture));
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(
            pipePair,
            new WorkerClientOptions
            {
                HeartbeatGrace = TimeSpan.FromMilliseconds(80),
                HeartbeatCheckInterval = TimeSpan.FromMilliseconds(20),
                EventChannelCapacity = 8,
                HeartbeatStuckCeiling = TimeSpan.FromMilliseconds(200),
            },
            timeProvider: clock);
        await CompleteHandshakeAsync(client, pipePair);

        Task<WorkerCommandReply> pendingInvoke = client.InvokeAsync(
            CreateCommand(MxCommandKind.Ping),
            TestTimeout,
            CancellationToken.None);
        await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);

        // Advance the clock past HeartbeatStuckCeiling. The worker pipe's
        // PendingCommand.StartTimestamp uses TimeProvider.GetTimestamp(), so the
        // ManualTimeProvider's GetElapsedTime sees the advanced gap.
        clock.Advance(TimeSpan.FromSeconds(2));

        await WaitUntilAsync(
            () => client.State == WorkerClientState.Faulted,
            TestTimeout);

        Assert.Equal(WorkerClientState.Faulted, client.State);
    }

    /// <summary>
    ///     Server-032 regression: a transient burst that exceeds
    ///     <see cref="WorkerClientOptions.EventChannelCapacity"/> must be
    ///     absorbed for up to <see cref="WorkerClientOptions.EventChannelFullModeTimeout"/>
    ///     (the channel is configured for <c>BoundedChannelFullMode.Wait</c>);
    ///     only when the wait elapses without progress is the worker faulted,
    ///     and the diagnostic must name the channel capacity, depth, and
    ///     actionable remediation.
    /// </summary>
    [Fact]
    public async Task EnqueueWorkerEvent_WhenChannelFullPastTimeout_FaultsWithRichDiagnostic()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(
            pipePair,
            new WorkerClientOptions
            {
                EventChannelCapacity = 4,
                EventChannelFullModeTimeout = TimeSpan.FromMilliseconds(100),
                HeartbeatGrace = TimeSpan.FromSeconds(30),
                HeartbeatCheckInterval = TimeSpan.FromSeconds(1),
            });
        await CompleteHandshakeAsync(client, pipePair);

        // Fill the 4-slot channel and write exactly one more to force the
        // overflow path. The gateway never opens a StreamEvents consumer, so
        // the events stay buffered. Exactly five events are written: the
        // worker client faults while reading the fifth, after which its read
        // loop stops — a sixth event would never be drained and its pipe
        // write would block forever on a full OS pipe buffer.
        for (ulong sequence = 1; sequence <= 5; sequence++)
        {
            await pipePair.WorkerWriter.WriteAsync(
                CreateEventEnvelope(sequence: sequence, MxEventFamily.OnDataChange));
        }

        await WaitUntilAsync(
            () => client.State == WorkerClientState.Faulted,
            TestTimeout);

        Assert.Equal(WorkerClientState.Faulted, client.State);

        // Reading the events channel after fault throws the propagated
        // WorkerClientException carrying the rich diagnostic message. The
        // drain is bounded by TestTimeout so a regression that leaves the
        // channel uncompleted fails the test instead of hanging it.
        using CancellationTokenSource drainTimeout = new(TestTimeout);
        WorkerClientException fault = await Assert.ThrowsAsync<WorkerClientException>(async () =>
        {
            await foreach (WorkerEvent _ in client.ReadEventsAsync(drainTimeout.Token))
            {
            }
        });
        Assert.Contains("Worker event channel rejected", fault.Message);
        Assert.Contains("of 4 capacity", fault.Message);
        Assert.Contains("StreamEvents", fault.Message);
        Assert.Contains("MxGateway:Events:QueueCapacity", fault.Message);
    }

    private static WorkerClient CreateClient(
        PipePair pipePair,
        WorkerClientOptions? options = null,
        GatewayMetrics? metrics = null,
        WorkerProcessHandle? processHandle = null,
        TimeProvider? timeProvider = null)
    {
        WorkerFrameProtocolOptions frameOptions = new(SessionId);
        WorkerClientConnection connection = new(
            SessionId,
            Nonce,
            pipePair.GatewayStream,
            frameOptions,
            processHandle);

        return new WorkerClient(connection, options, metrics, timeProvider);
    }

    private static WorkerProcessHandle CreateProcessHandle(FakeWorkerProcess process)
    {
        return new WorkerProcessHandle(
            process,
            new WorkerProcessCommandLine("ZB.MOM.WW.MxGateway.Worker.exe", []),
            DateTimeOffset.UtcNow);
    }

    private static async Task CompleteHandshakeAsync(
        WorkerClient client,
        PipePair pipePair)
    {
        Task startTask = client.StartAsync(CancellationToken.None);

        WorkerEnvelope gatewayHello = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
        Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, gatewayHello.BodyCase);
        Assert.Equal(Nonce, gatewayHello.GatewayHello.Nonce);
        Assert.Equal(GatewayContractInfo.WorkerProtocolVersion, gatewayHello.GatewayHello.SupportedProtocolVersion);

        await pipePair.WorkerWriter.WriteAsync(CreateWorkerHelloEnvelope());
        await pipePair.WorkerWriter.WriteAsync(CreateWorkerReadyEnvelope());
        await startTask.WaitAsync(TestTimeout);
    }

    private static WorkerCommand CreateCommand(MxCommandKind kind)
    {
        return new WorkerCommand
        {
            Command = new MxCommand
            {
                Kind = kind,
            },
        };
    }

    private static WorkerEnvelope CreateWorkerHelloEnvelope()
    {
        return CreateWorkerEnvelope(
            correlationId: string.Empty,
            sequence: 1,
            envelope => envelope.WorkerHello = new WorkerHello
            {
                ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
                Nonce = Nonce,
                WorkerProcessId = WorkerProcessId,
                WorkerVersion = "fake-worker",
            });
    }

    private static WorkerEnvelope CreateWorkerReadyEnvelope()
    {
        return CreateWorkerEnvelope(
            correlationId: string.Empty,
            sequence: 2,
            envelope => envelope.WorkerReady = new WorkerReady
            {
                WorkerProcessId = WorkerProcessId,
                MxaccessProgid = "LMXProxy.LMXProxyServer.1",
                MxaccessClsid = "{C30B52F5-2CB5-4760-AF0A-3A344A7EB5DC}",
            });
    }

    private static WorkerEnvelope CreateCommandReplyEnvelope(
        string correlationId,
        MxCommandKind kind)
    {
        return CreateWorkerEnvelope(
            correlationId,
            sequence: 10,
            envelope => envelope.WorkerCommandReply = new WorkerCommandReply
            {
                Reply = new MxCommandReply
                {
                    SessionId = SessionId,
                    CorrelationId = correlationId,
                    Kind = kind,
                },
            });
    }

    private static WorkerEnvelope CreateEventEnvelope(
        ulong sequence,
        MxEventFamily family)
    {
        return CreateWorkerEnvelope(
            correlationId: string.Empty,
            sequence,
            envelope => envelope.WorkerEvent = new WorkerEvent
            {
                Event = new MxEvent
                {
                    SessionId = SessionId,
                    Family = family,
                    WorkerSequence = sequence,
                },
            });
    }

    private static WorkerEnvelope CreateWorkerFaultEnvelope(string diagnosticMessage)
    {
        return CreateWorkerEnvelope(
            correlationId: string.Empty,
            sequence: 30,
            envelope => envelope.WorkerFault = new WorkerFault
            {
                Category = WorkerFaultCategory.MxaccessCommandFailed,
                DiagnosticMessage = diagnosticMessage,
                ProtocolStatus = new ProtocolStatus
                {
                    Code = ProtocolStatusCode.WorkerUnavailable,
                    Message = diagnosticMessage,
                },
            });
    }

    private static WorkerEnvelope CreateHeartbeatEnvelope(int workerProcessId)
    {
        return CreateWorkerEnvelope(
            correlationId: string.Empty,
            sequence: 20,
            envelope => envelope.WorkerHeartbeat = new WorkerHeartbeat
            {
                WorkerProcessId = workerProcessId,
                State = WorkerState.Ready,
                LastStaActivityTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
                PendingCommandCount = 0,
                OutboundEventQueueDepth = 0,
            });
    }

    private static WorkerEnvelope CreateWorkerEnvelope(
        string correlationId,
        ulong sequence,
        Action<WorkerEnvelope> setBody)
    {
        WorkerEnvelope envelope = new()
        {
            ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
            SessionId = SessionId,
            Sequence = sequence,
            CorrelationId = correlationId,
        };
        setBody(envelope);

        return envelope;
    }

    private static async Task WaitUntilAsync(
        Func<bool> predicate,
        TimeSpan timeout)
    {
        using CancellationTokenSource cancellationTokenSource = new(timeout);
        while (!predicate())
        {
            await Task.Delay(TimeSpan.FromMilliseconds(10), cancellationTokenSource.Token);
        }
    }

    private sealed class PipePair : IAsyncDisposable
    {
        private readonly NamedPipeClientStream _workerStream;
        private bool _workerSideDisposed;

        private PipePair(
            NamedPipeServerStream gatewayStream,
            NamedPipeClientStream workerStream)
        {
            GatewayStream = gatewayStream;
            _workerStream = workerStream;
            WorkerReader = new WorkerFrameReader(_workerStream, new WorkerFrameProtocolOptions(SessionId));
            WorkerWriter = new WorkerFrameWriter(_workerStream, new WorkerFrameProtocolOptions(SessionId));
        }

        /// <summary>The gateway side of the named pipe connection.</summary>
        public NamedPipeServerStream GatewayStream { get; }

        /// <summary>Frame reader for worker messages.</summary>
        public WorkerFrameReader WorkerReader { get; }

        /// <summary>Frame writer for worker messages.</summary>
        public WorkerFrameWriter WorkerWriter { get; }

        /// <summary>Creates a connected pipe pair for testing.</summary>
        public static async Task<PipePair> CreateAsync()
        {
            string pipeName = $"mxaccessgw-workerclient-tests-{Guid.NewGuid():N}";
            NamedPipeServerStream gatewayStream = new(
                pipeName,
                PipeDirection.InOut,
                maxNumberOfServerInstances: 1,
                PipeTransmissionMode.Byte,
                PipeOptions.Asynchronous);
            NamedPipeClientStream workerStream = new(
                ".",
                pipeName,
                PipeDirection.InOut,
                PipeOptions.Asynchronous);

            Task waitForConnectionTask = gatewayStream.WaitForConnectionAsync();
            await workerStream.ConnectAsync();
            await waitForConnectionTask;

            return new PipePair(gatewayStream, workerStream);
        }

        /// <summary>Disposes the worker side of the pipe.</summary>
        public async ValueTask DisposeWorkerSideAsync()
        {
            if (_workerSideDisposed)
            {
                return;
            }

            await _workerStream.DisposeAsync();
            _workerSideDisposed = true;
        }

        /// <summary>Disposes the duplex stream.</summary>
        public async ValueTask DisposeAsync()
        {
            await DisposeWorkerSideAsync();
            await GatewayStream.DisposeAsync();
        }
    }

    private sealed class FakeWorkerProcess : IWorkerProcess
    {
        private readonly TaskCompletionSource _exited = new(TaskCreationOptions.RunContinuationsAsynchronously);

        /// <summary>Gets the process ID.</summary>
        public int Id { get; } = WorkerProcessId;

        /// <summary>Gets a value indicating whether the process has exited.</summary>
        public bool HasExited { get; private set; }

        /// <summary>Gets the process exit code.</summary>
        public int? ExitCode { get; private set; }

        /// <summary>Gets the number of times kill was called.</summary>
        public int KillCount { get; private set; }

        /// <summary>Gets the last kill request's entire process tree flag.</summary>
        public bool KillEntireProcessTree { get; private set; }

        /// <summary>Gets a value indicating whether dispose was called.</summary>
        public bool Disposed { get; private set; }

        /// <inheritdoc />
        public ValueTask WaitForExitAsync(CancellationToken cancellationToken)
        {
            return new ValueTask(_exited.Task.WaitAsync(cancellationToken));
        }

        /// <summary>Records a kill request.</summary>
        /// <param name="entireProcessTree">Whether to kill the entire process tree.</param>
        public void Kill(bool entireProcessTree)
        {
            KillCount++;
            KillEntireProcessTree = entireProcessTree;
            HasExited = true;
            ExitCode = -1;
            _exited.TrySetResult();
        }

        /// <inheritdoc />
        public void Dispose()
        {
            Disposed = true;
        }
    }
}