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.Gateway.Workers.Fakes;
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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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>
    ///     The <c>GatewayHello</c> carries the negotiated worker-frame maximum so the worker adopts the
    ///     configured limit instead of its own default (IPC-02); a regression to sending <c>0</c> means
    ///     "older gateway, use default" to the worker and would silently downgrade the negotiated limit.
    ///     The adoption half is asserted only in the Windows-only worker suite, so the gateway half is
    ///     pinned here, in the portable suite (TST-28). Both the default and an override are covered so
    ///     the assertion tracks configuration rather than a constant.
    /// </summary>
    /// <param name="maxMessageBytes">Configured worker-frame maximum to negotiate.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Theory]
    [InlineData(WorkerFrameProtocolOptions.DefaultMaxMessageBytes)]
    [InlineData(2 * 1024 * 1024)]
    public async Task StartAsync_SendsGatewayHelloWithConfiguredMaxFrameBytes(int maxMessageBytes)
    {
        await using FakeWorkerHarness harness =
            await FakeWorkerHarness.CreateConnectedPairAsync(maxMessageBytes: maxMessageBytes);
        await using WorkerClient client = harness.CreateClient();

        Task startTask = client.StartAsync(CancellationToken.None);
        WorkerEnvelope gatewayHello = await harness.CompleteStartupAsync().WaitAsync(TestTimeout);
        await startTask.WaitAsync(TestTimeout);

        Assert.NotEqual(0u, gatewayHello.GatewayHello.MaxFrameBytes);
        Assert.Equal((uint)maxMessageBytes, gatewayHello.GatewayHello.MaxFrameBytes);
    }

    /// <summary>Verifies that InvokeAsync completes a pending command when a matching reply arrives.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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.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 a command whose serialized envelope exceeds the negotiated worker-frame maximum
    /// fails only that command with <see cref="WorkerClientErrorCode.CommandTooLarge"/> at the enqueue
    /// boundary, leaving the client ready for subsequent commands. Without the pre-check the
    /// oversized frame would reach the write loop and fault the whole session.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task InvokeAsync_WhenCommandExceedsFrameMax_FailsOnlyThatCommandAndStaysReady()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair, maxMessageBytes: 4096);
        await CompleteHandshakeAsync(client, pipePair);

        WorkerCommand oversized = new()
        {
            Command = new MxCommand
            {
                Kind = MxCommandKind.Write,
                Write = new WriteCommand
                {
                    ServerHandle = 1,
                    ItemHandle = 2,
                    Value = new MxValue { StringValue = new string('x', 8192) },
                },
            },
        };

        WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
            async () => await client.InvokeAsync(oversized, TestTimeout, CancellationToken.None));
        Assert.Equal(WorkerClientErrorCode.CommandTooLarge, exception.ErrorCode);
        Assert.Equal(WorkerClientState.Ready, client.State);

        // A subsequent normally-sized command still round-trips: the session was not faulted.
        Task<WorkerCommandReply> nextInvoke = client.InvokeAsync(
            CreateCommand(MxCommandKind.Ping),
            TestTimeout,
            CancellationToken.None);
        WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
        await pipePair.WriteAsync(
            CreateCommandReplyEnvelope(commandEnvelope.CorrelationId, MxCommandKind.Ping));
        WorkerCommandReply reply = await nextInvoke.WaitAsync(TestTimeout);
        Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind);
        Assert.Equal(WorkerClientState.Ready, client.State);
    }

    /// <summary>Verifies that InvokeAsync ignores late replies and keeps the client ready.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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.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.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>
    ///     The envelope <c>sequence</c> is a monotonic per-sender counter (gateway.md), so the values
    ///     observed on the pipe must be strictly increasing in wire order. Stamping the sequence when
    ///     the envelope is constructed lets two concurrent invokes stamp 1 and 2 but enqueue 2 then 1
    ///     (GWC-28); stamping on the single-consumer write loop makes wire order and sequence order the
    ///     same thing by construction.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConcurrentInvokesEmitStrictlyIncreasingSequencesOnTheWire()
    {
        const int commandCount = 32;
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);

        Task<WorkerCommandReply>[] invokeTasks = Enumerable.Range(0, commandCount)
            .Select(_ => Task.Run(async () => await client.InvokeAsync(
                CreateCommand(MxCommandKind.Ping),
                TestTimeout,
                CancellationToken.None)))
            .ToArray();

        ulong previousSequence = 0;
        for (int index = 0; index < commandCount; index++)
        {
            WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
            Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase);
            Assert.True(
                commandEnvelope.Sequence > previousSequence,
                $"Command {index} arrived with sequence {commandEnvelope.Sequence} after {previousSequence}; "
                + "envelope sequences must be strictly increasing in wire order.");
            previousSequence = commandEnvelope.Sequence;

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

        await Task.WhenAll(invokeTasks).WaitAsync(TestTimeout);
        Assert.Equal(WorkerClientState.Ready, client.State);
    }

    /// <summary>Verifies that ReadEventsAsync yields events in pipe order from the worker.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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.WriteAsync(
            CreateEventEnvelope(sequence: 11, MxEventFamily.OnDataChange));
        await pipePair.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>
    ///     The worker event channel is single-reader: a second <see cref="WorkerClient.ReadEventsAsync"/>
    ///     enumerator must throw rather than silently split events between two consumers.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadEventsAsync_SecondEnumerator_Throws()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);
        using CancellationTokenSource cancellationTokenSource = new(TestTimeout);

        // First call claims the single reader.
        IAsyncEnumerable<WorkerEvent> first = client.ReadEventsAsync(cancellationTokenSource.Token);
        await using IAsyncEnumerator<WorkerEvent> firstEnumerator = first.GetAsyncEnumerator(cancellationTokenSource.Token);

        // A second call must fail loudly at call time rather than racing the first for events.
        Assert.Throws<InvalidOperationException>(() => client.ReadEventsAsync(cancellationTokenSource.Token));
    }

    /// <summary>Verifies that the read loop faults the client when the event queue overflows.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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.WriteAsync(
            CreateEventEnvelope(sequence: 11, MxEventFamily.OnDataChange));
        await pipePair.WriteAsync(
            CreateEventEnvelope(sequence: 12, MxEventFamily.OnDataChange));

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

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

    /// <summary>
    /// Verifies that a command reply arriving on the pipe after events is dispatched promptly even
    /// when the event channel is full and has no consumer — event enqueue is decoupled from the read
    /// loop, so a blocked event writer cannot delay a reply. The event full-mode timeout is
    /// set far above the command timeout: without the decoupling the read loop would block behind the
    /// full event channel and the in-flight InvokeAsync would hit CommandTimeout.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadLoop_WhenEventChannelFull_DispatchesCommandReplyWithoutBlocking()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(
            pipePair,
            new WorkerClientOptions
            {
                EventChannelCapacity = 1,
                // Much larger than TestTimeout: the read loop must not block for this long behind the
                // full event channel while a reply waits.
                EventChannelFullModeTimeout = TimeSpan.FromSeconds(30),
                HeartbeatGrace = TimeSpan.FromSeconds(60),
                HeartbeatCheckInterval = TimeSpan.FromSeconds(60),
            });
        await CompleteHandshakeAsync(client, pipePair);

        // No StreamEvents consumer is attached, so the capacity-1 event channel fills and the event
        // writer blocks on the second event's timed WriteAsync.
        await pipePair.WriteAsync(
            CreateEventEnvelope(sequence: 11, MxEventFamily.OnDataChange));
        await pipePair.WriteAsync(
            CreateEventEnvelope(sequence: 12, MxEventFamily.OnDataChange));

        Task<WorkerCommandReply> invokeTask = client.InvokeAsync(
            CreateCommand(MxCommandKind.Ping),
            TestTimeout,
            CancellationToken.None);
        WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
        await pipePair.WriteAsync(
            CreateCommandReplyEnvelope(commandEnvelope.CorrelationId, MxCommandKind.Ping));

        WorkerCommandReply reply = await invokeTask.WaitAsync(TestTimeout);
        Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind);
        Assert.Equal(WorkerClientState.Ready, 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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadLoop_WhenClientFaults_KillsOwnedWorkerProcess()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        FakeWorkerProcess process = new(WorkerProcessId);
        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.WriteAsync(
            CreateEventEnvelope(sequence: 11, MxEventFamily.OnDataChange));
        await pipePair.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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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.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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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 DisposeAsync kills the still-running owned worker process before disposing.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DisposeAsync_WhenOwnedWorkerStillRuns_KillsProcessBeforeDisposing()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        FakeWorkerProcess process = new(WorkerProcessId);
        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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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.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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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>
    ///     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.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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>
    ///     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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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>
    ///     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>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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.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);
    }

    /// <summary>
    ///     GWC-24: the staging channel between the read loop and the event writer is bounded at
    ///     2 × <see cref="WorkerClientOptions.EventChannelCapacity"/>, so a consumer that drains
    ///     slower than the worker produces faults the client at the bound instead of growing
    ///     gateway memory silently. The full-mode timeout here is far longer than the test could
    ///     ever wait, which proves the fault came from the staging bound and not from the timed
    ///     write in <c>EnqueueWorkerEventAsync</c>. A command reply interleaved before the fault
    ///     must still complete: the read loop never blocks behind events (the GWC-04 guarantee).
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StagingChannelOverflowFaultsWorkerWithoutWaitingForFullModeTimeout()
    {
        const int capacity = 4;
        const int stagingBound = 2 * capacity;
        using GatewayMetrics metrics = new();
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(
            pipePair,
            new WorkerClientOptions
            {
                EventChannelCapacity = capacity,
                // Five minutes: the timed-write fault physically cannot be the trigger.
                EventChannelFullModeTimeout = TimeSpan.FromMinutes(5),
                HeartbeatGrace = TimeSpan.FromSeconds(30),
                HeartbeatCheckInterval = TimeSpan.FromSeconds(30),
            },
            metrics: metrics);
        await CompleteHandshakeAsync(client, pipePair);

        // The staging channel alone holds 2 × capacity, and the event writer only ever removes
        // from it, so this batch cannot fault however the writer happens to be scheduled. Waiting
        // on the gauge (rather than a delay) proves the read loop consumed every one of them.
        ulong sequence = 1;
        for (; sequence <= (ulong)stagingBound; sequence++)
        {
            await pipePair.WriteAsync(
                CreateEventEnvelope(sequence, MxEventFamily.OnDataChange));
        }

        await WaitUntilAsync(
            () => metrics.GetSnapshot().WorkerEventQueueDepth == stagingBound,
            TestTimeout);
        Assert.Equal(WorkerClientState.Ready, client.State);

        // The event path is backed up with no consumer attached, yet a command still round-trips.
        Task<WorkerCommandReply> invokeTask = client.InvokeAsync(
            CreateCommand(MxCommandKind.Ping),
            TestTimeout,
            CancellationToken.None);
        WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
        await pipePair.WriteAsync(
            CreateCommandReplyEnvelope(commandEnvelope.CorrelationId, MxCommandKind.Ping));
        WorkerCommandReply reply = await invokeTask.WaitAsync(TestTimeout);
        Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind);
        Assert.Equal(WorkerClientState.Ready, client.State);

        // The absolute ceiling is capacity (consumer channel) + 1 (in flight in the blocked event
        // writer) + 2 × capacity (staging). Push comfortably past it. The events are tiny, so the
        // ones the stopped read loop never drains stay in the OS pipe buffer instead of blocking.
        for (int extra = 0; extra < 3 * capacity; extra++, sequence++)
        {
            await pipePair.WriteAsync(
                CreateEventEnvelope(sequence, MxEventFamily.OnDataChange));
        }

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

        using CancellationTokenSource drainTimeout = new(TestTimeout);
        WorkerClientException fault = await Assert.ThrowsAsync<WorkerClientException>(async () =>
        {
            await foreach (WorkerEvent _ in client.ReadEventsAsync(drainTimeout.Token))
            {
            }
        });

        Assert.Equal(WorkerClientErrorCode.ProtocolViolation, fault.ErrorCode);
        Assert.Contains("staging", fault.Message, StringComparison.OrdinalIgnoreCase);
        Assert.Contains($"{2 * capacity}", fault.Message, StringComparison.Ordinal);
        Assert.Contains("StreamEvents", fault.Message, StringComparison.Ordinal);
        Assert.Contains("MxGateway:Events:QueueCapacity", fault.Message, StringComparison.Ordinal);

        // The timed-write diagnostic must not be what fired.
        Assert.DoesNotContain("Worker event channel rejected", fault.Message, StringComparison.Ordinal);
    }

    /// <summary>
    ///     GWC-24: the worker event queue-depth gauge counts staged *and* queued events, so a
    ///     backlog held in the staging channel is visible rather than invisible. Depth is
    ///     incremented at staging and decremented when the consumer reads, so the single counter
    ///     reports total undelivered events and returns to zero once drained.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WorkerEventQueueDepthGaugeCountsStagedEvents()
    {
        const int capacity = 4;
        const int eventCount = 8;
        using GatewayMetrics metrics = new();
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(
            pipePair,
            new WorkerClientOptions
            {
                EventChannelCapacity = capacity,
                EventChannelFullModeTimeout = TimeSpan.FromMinutes(5),
                HeartbeatGrace = TimeSpan.FromSeconds(30),
                HeartbeatCheckInterval = TimeSpan.FromSeconds(30),
            },
            metrics: metrics);
        await CompleteHandshakeAsync(client, pipePair);

        // Above EventChannelCapacity but below the 2× staging bound: no consumer, no fault.
        for (ulong sequence = 1; sequence <= eventCount; sequence++)
        {
            await pipePair.WriteAsync(
                CreateEventEnvelope(sequence, MxEventFamily.OnDataChange));
        }

        await WaitUntilAsync(
            () => metrics.GetSnapshot().WorkerEventQueueDepth == eventCount,
            TestTimeout);
        Assert.Equal(eventCount, metrics.GetSnapshot().WorkerEventQueueDepth);
        Assert.Equal(WorkerClientState.Ready, client.State);

        using CancellationTokenSource drainTimeout = new(TestTimeout);
        await using IAsyncEnumerator<WorkerEvent> events = client
            .ReadEventsAsync(drainTimeout.Token)
            .GetAsyncEnumerator(drainTimeout.Token);
        for (int read = 0; read < eventCount; read++)
        {
            Assert.True(await events.MoveNextAsync());
        }

        await WaitUntilAsync(
            () => metrics.GetSnapshot().WorkerEventQueueDepth == 0,
            TestTimeout);
        Assert.Equal(0, metrics.GetSnapshot().WorkerEventQueueDepth);
    }

    private static WorkerClient CreateClient(
        PipePair pipePair,
        WorkerClientOptions? options = null,
        GatewayMetrics? metrics = null,
        WorkerProcessHandle? processHandle = null,
        TimeProvider? timeProvider = null,
        int maxMessageBytes = WorkerFrameProtocolOptions.DefaultMaxMessageBytes)
    {
        WorkerFrameProtocolOptions frameOptions = new(
            SessionId,
            GatewayContractInfo.WorkerProtocolVersion,
            maxMessageBytes);
        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.WriteAsync(CreateWorkerHelloEnvelope());
        await pipePair.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>
        /// Writes one envelope from the fake worker side, bounded by <see cref="TestTimeout"/>.
        /// </summary>
        /// <remarks>
        /// Every test-side write goes through here rather than <see cref="WorkerWriter"/> directly so
        /// that a write which cannot complete fails this test instead of hanging it. An unbounded
        /// write is not a local problem: a test method that never returns leaves xUnit's assembly
        /// runner awaiting it forever, so <c>ITestAssemblyFinished</c> is never raised and the whole
        /// <c>testhost</c> process never exits — the run reports no failure at all, just a wedge.
        /// </remarks>
        /// <param name="envelope">The envelope to write.</param>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public async Task WriteAsync(WorkerEnvelope envelope)
        {
            using CancellationTokenSource writeTimeout = new(TestTimeout);
            try
            {
                await WorkerWriter.WriteAsync(envelope, writeTimeout.Token);
            }
            catch (OperationCanceledException) when (writeTimeout.IsCancellationRequested)
            {
                Assert.Fail(
                    $"The fake worker's pipe write did not complete within {TestTimeout}. The gateway " +
                    "side has stopped reading and the pipe buffer is full.");
            }
        }

        /// <summary>Creates a connected pipe pair for testing.</summary>
        /// <returns>The connected <see cref="PipePair"/>.</returns>
        public static async Task<PipePair> CreateAsync()
        {
            string pipeName = $"mxgw-wc-{Guid.NewGuid():N}";
            NamedPipeServerStream gatewayStream = TestNamedPipe.CreateServer(pipeName);
            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>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public async ValueTask DisposeWorkerSideAsync()
        {
            if (_workerSideDisposed)
            {
                return;
            }

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

        /// <summary>Disposes the duplex stream.</summary>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public async ValueTask DisposeAsync()
        {
            await DisposeWorkerSideAsync();
            await GatewayStream.DisposeAsync();
        }
    }

}