mxaccessgw/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs

using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Metrics;

namespace ZB.MOM.WW.MxGateway.Server.Workers;

public sealed class WorkerClient : IWorkerClient
{
    private const string GatewayVersionFallback = "unknown";
    private static readonly TimeSpan DisposeTaskTimeout = TimeSpan.FromSeconds(5);
    private readonly object _syncRoot = new();
    private readonly WorkerClientConnection _connection;
    private readonly WorkerClientOptions _options;
    private readonly GatewayMetrics? _metrics;
    private readonly TimeProvider _timeProvider;
    private readonly ILogger<WorkerClient> _logger;
    private readonly WorkerFrameReader _reader;
    private readonly WorkerFrameWriter _writer;
    private readonly Channel<WorkerEnvelope> _outboundEnvelopes;
    private readonly Channel<WorkerEvent> _events;

    // Staging hand-off between the read loop and the dedicated event writer. The read loop writes
    // here with a non-blocking TryWrite so a full consumer channel (_events) can never stall the read
    // loop behind an event — replies and heartbeats keep flowing. Bounded at 2 × EventChannelCapacity
    // because an unbounded staging channel let a consumer that drains slower than the worker
    // produces grow gateway memory without limit and without a fault, because each individual timed
    // write into _events still completed inside EventChannelFullModeTimeout. A rejected TryWrite here
    // is the sustained-slow-drain signal and faults the client immediately (ProtocolViolation), which
    // is the fail-fast backpressure policy in docs/DesignDecisions.md. Total gateway-side buffering
    // per session is therefore 3 × MxGateway:Events:QueueCapacity, and the single _eventQueueDepth
    // gauge covers staged + queued events so the whole backlog is observable.
    private readonly Channel<WorkerEvent> _eventStaging;
    private readonly int _eventStagingCapacity;
    private readonly ConcurrentDictionary<string, PendingCommand> _pendingCommands = new(StringComparer.Ordinal);
    private readonly SemaphoreSlim _pendingCommandSlots;
    private readonly CancellationTokenSource _stopCts = new();

    // this client's contribution to the gateway-wide worker queue-depth gauge. Registered
    // once here and read only when the gauge is scraped, so staging and consuming an event cost an
    // Interlocked on _eventQueueDepth and nothing else — previously each of those two hot-path steps
    // called into GatewayMetrics and took its process-wide lock. Null when metrics are disabled.
    private readonly IDisposable? _eventQueueDepthRegistration;
    // Touched only by WriteLoopAsync — the single consumer of _outboundEnvelopes — so it needs no
    // interlocking. See WriteLoopAsync for why the stamp happens there rather than at construction.
    private ulong _nextSequence;
    private WorkerClientState _state;
    private DateTimeOffset _lastHeartbeatAt;
    private int? _processId;
    private int _eventQueueDepth;
    private int _eventsReaderClaimed;
    private Task? _readLoopTask;
    private Task? _writeLoopTask;
    private Task? _eventWriteLoopTask;
    private Task? _heartbeatLoopTask;
    private bool _workerStartRecorded;
    private bool _disposed;

    /// <summary>Initializes a client for communicating with a worker process over a named pipe.</summary>
    /// <param name="connection">Named pipe connection to the worker.</param>
    /// <param name="options">Worker client configuration; defaults to WorkerClientOptions if null.</param>
    /// <param name="metrics">Gateway metrics sink; null disables metrics recording.</param>
    /// <param name="timeProvider">Time provider for timestamps; defaults to system time if null.</param>
    /// <param name="logger">Logger instance; defaults to NullLogger if null.</param>
    public WorkerClient(
        WorkerClientConnection connection,
        WorkerClientOptions? options = null,
        GatewayMetrics? metrics = null,
        TimeProvider? timeProvider = null,
        ILogger<WorkerClient>? logger = null)
    {
        _connection = connection ?? throw new ArgumentNullException(nameof(connection));
        _options = options ?? new WorkerClientOptions();
        _metrics = metrics;
        _timeProvider = timeProvider ?? TimeProvider.System;
        _logger = logger ?? NullLogger<WorkerClient>.Instance;
        _reader = new WorkerFrameReader(connection.Stream, connection.FrameOptions);
        _writer = new WorkerFrameWriter(connection.Stream, connection.FrameOptions);
        _pendingCommandSlots = new SemaphoreSlim(_options.MaxPendingCommands, _options.MaxPendingCommands);
        _outboundEnvelopes = Channel.CreateBounded<WorkerEnvelope>(
            new BoundedChannelOptions(_options.MaxPendingCommands + 4)
            {
                SingleReader = true,
                SingleWriter = false,
                FullMode = BoundedChannelFullMode.Wait,
                AllowSynchronousContinuations = false,
            });
        _events = Channel.CreateBounded<WorkerEvent>(
            new BoundedChannelOptions(_options.EventChannelCapacity)
            {
                // The worker event channel has exactly ONE consumer: the per-session
                // SessionEventDistributor pump. The alarm monitor and dashboard mirror both
                // attach to the distributor rather than draining this channel directly, so a
                // second concurrent reader would silently split events between the two
                // enumerators. SingleReader=true asserts that invariant; ReadEventsAsync adds a
                // claimed-once guard so a regression fails loudly instead of losing events.
                SingleReader = true,
                SingleWriter = true,
                FullMode = BoundedChannelFullMode.Wait,
                AllowSynchronousContinuations = false,
            });
        _eventStagingCapacity = checked(2 * _options.EventChannelCapacity);
        _eventStaging = Channel.CreateBounded<WorkerEvent>(
            new BoundedChannelOptions(_eventStagingCapacity)
            {
                // The read loop is the only writer; EventWriteLoopAsync is the only reader.
                SingleReader = true,
                SingleWriter = true,
                // Wait (not Drop*) so the read loop's non-blocking TryWrite returns false exactly
                // when the bound is reached — the same Wait+TryWrite overflow-detection idiom the
                // session event distributor uses. The read loop never awaits this channel.
                FullMode = BoundedChannelFullMode.Wait,
                AllowSynchronousContinuations = false,
            });
        _eventQueueDepthRegistration = _metrics?.RegisterWorkerEventQueueDepthSource(
            () => Volatile.Read(ref _eventQueueDepth));
        _lastHeartbeatAt = _timeProvider.GetUtcNow();
    }

    /// <inheritdoc />
    public string SessionId => _connection.SessionId;

    /// <inheritdoc />
    public int? ProcessId
    {
        get
        {
            lock (_syncRoot)
            {
                return _processId;
            }
        }
    }

    /// <inheritdoc />
    public WorkerClientState State
    {
        get
        {
            lock (_syncRoot)
            {
                return _state;
            }
        }
    }

    /// <inheritdoc />
    public DateTimeOffset LastHeartbeatAt
    {
        get
        {
            lock (_syncRoot)
            {
                return _lastHeartbeatAt;
            }
        }
    }

    /// <inheritdoc />
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        ThrowIfDisposed();
        TransitionFromCreatedToHandshaking();

        _writeLoopTask = Task.Run(WriteLoopAsync);
        await EnqueueAsync(CreateGatewayHelloEnvelope(), cancellationToken).ConfigureAwait(false);

        WorkerEnvelope helloEnvelope = await ReadHandshakeEnvelopeAsync(
            WorkerEnvelope.BodyOneofCase.WorkerHello,
            cancellationToken).ConfigureAwait(false);
        ValidateWorkerHello(helloEnvelope.WorkerHello);

        WorkerEnvelope readyEnvelope = await ReadHandshakeEnvelopeAsync(
            WorkerEnvelope.BodyOneofCase.WorkerReady,
            cancellationToken).ConfigureAwait(false);
        MarkReady(readyEnvelope.WorkerReady);

        _readLoopTask = Task.Run(ReadLoopAsync);
        _eventWriteLoopTask = Task.Run(EventWriteLoopAsync);
        _heartbeatLoopTask = Task.Run(HeartbeatLoopAsync);
    }

    /// <inheritdoc />
    public async Task<WorkerCommandReply> InvokeAsync(
        WorkerCommand command,
        TimeSpan timeout,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(command);
        ThrowIfDisposed();
        EnsureReady();

        if (timeout <= TimeSpan.Zero)
        {
            throw new ArgumentOutOfRangeException(nameof(timeout), timeout, "Command timeout must be greater than zero.");
        }

        string correlationId = Guid.NewGuid().ToString("N");
        string method = GetCommandMethod(command);
        if (!_pendingCommandSlots.Wait(0))
        {
            _metrics?.QueueOverflow("worker-pending-commands");
            throw new WorkerClientException(
                WorkerClientErrorCode.PendingCommandLimitExceeded,
                $"Worker session {SessionId} already has {_options.MaxPendingCommands} pending command(s).");
        }

        PendingCommand pendingCommand = new(
            correlationId,
            method,
            _timeProvider.GetTimestamp());

        if (!_pendingCommands.TryAdd(correlationId, pendingCommand))
        {
            ReleasePendingCommandSlot();
            throw new InvalidOperationException("Generated a duplicate command correlation id.");
        }

        _metrics?.CommandStarted(method);

        try
        {
            WorkerEnvelope commandEnvelope = CreateCommandEnvelope(correlationId, command);

            // Reject an oversized command at the enqueue boundary so only this correlation fails
            // (ResourceExhausted) rather than the frame reaching the write loop and faulting the whole
            // session. Command envelopes are the only gateway-authored outbound payload whose
            // size the caller controls; checking here keeps a MessageTooLarge in the write loop a
            // genuine desync signal.
            //
            // this size cannot be handed to WorkerFrameWriter to spare its own
            // CalculateSize. WriteLoopAsync stamps envelope.Sequence immediately before the write
            // and a non-zero varint field grows the encoding — so the number computed here
            // is a lower bound on the frame the writer actually emits, never the frame length. Passing
            // it as a knownSize would under-length the prefix and desync the worker's framing.
            //
            // being a lower bound is conservative for the knownSize question but the wrong
            // direction for this pre-check, which must not admit an envelope the writer will then
            // reject with a session-fatal MessageTooLarge. Compare against the frame maximum less the
            // MaxSequenceEncodingBytes the stamp can add, so every envelope this check accepts still
            // fits once the sequence is written.
            int envelopeSize = commandEnvelope.CalculateSize();
            int preCheckLimit = _connection.FrameOptions.MaxMessageBytes - MaxSequenceEncodingBytes;
            if (envelopeSize > preCheckLimit)
            {
                throw new WorkerClientException(
                    WorkerClientErrorCode.CommandTooLarge,
                    $"Worker command {method} serializes to {envelopeSize} bytes, exceeding the negotiated "
                    + $"worker-frame maximum of {_connection.FrameOptions.MaxMessageBytes} bytes less the "
                    + $"{MaxSequenceEncodingBytes}-byte sequence-stamp allowance.");
            }

            await EnqueueAsync(commandEnvelope, cancellationToken).ConfigureAwait(false);

            // one pooled timer instead of a linked CTS + Task.Delay + WhenAny per command.
            // Task.WaitAsync arms a TimerQueueTimer on the shared timer queue and cancels it when the
            // reply lands, so the steady-state cost of a command that replies in time is a single
            // continuation — the old shape allocated a linked CancellationTokenSource, its
            // registration, a delay Task, and the WhenAny Task on every invoke, and left the delay
            // Task rooted until the cancel completed. WaitAsync raises TimeoutException for the
            // deadline and OperationCanceledException for the caller's token — but unlike the old
            // wait it races the two and reports whichever fired first, whereas the old code inspected
            // cancellationToken.IsCancellationRequested BEFORE classifying a won delay as a timeout.
            // The filter on the CommandTimeout clause restores that priority: a token canceled around
            // the deadline is still classified as cancellation, never as CommandTimeout. Error codes
            // and messages are unchanged.
            try
            {
                return await pendingCommand.Task.WaitAsync(timeout, cancellationToken).ConfigureAwait(false);
            }
            catch (TimeoutException) when (!cancellationToken.IsCancellationRequested)
            {
                string timeoutMessage = $"Worker command {method} timed out after {timeout}.";
                bool removed = RemovePendingCommandAsFailed(
                    correlationId,
                    pendingCommand,
                    WorkerClientErrorCode.CommandTimeout,
                    timeoutMessage);

                // The gateway has stopped waiting, but the worker has not stopped working: the
                // correlation is still on its single STA queue and would execute (or keep executing)
                // regardless. Tell it, so a queued-but-not-started command is dropped instead of
                // occupying the STA behind a caller that is already gone. Gated on the removal so a
                // reply that won the race — the pending entry is already gone and the caller is about
                // to see it — never has a cancel chase it. Best-effort by design; the send cannot
                // throw, so it can never replace the timeout the caller is owed.
                if (removed)
                {
                    TrySendCancelForAbandonedCommand(
                        correlationId,
                        method,
                        $"gateway command timeout after {timeout}");
                }

                throw new WorkerClientException(
                    WorkerClientErrorCode.CommandTimeout,
                    timeoutMessage);
            }
            catch (OperationCanceledException)
            {
                AbandonCommandOnCallerCancellation(correlationId, pendingCommand, method);

                // WaitAsync surfaces TaskCanceledException; throwing through the token keeps the
                // exception the caller observes exactly what the hand-rolled wait produced.
                cancellationToken.ThrowIfCancellationRequested();
                throw;
            }
            catch (TimeoutException)
            {
                // The deadline and the caller's cancellation raced and WaitAsync picked the timer.
                // The old wait classified this as cancellation, so this clause — reached only when
                // the filter above saw a canceled token — reproduces that treatment exactly.
                AbandonCommandOnCallerCancellation(correlationId, pendingCommand, method);
                cancellationToken.ThrowIfCancellationRequested();
                throw;
            }
        }
        catch
        {
            if (_pendingCommands.TryRemove(correlationId, out _))
            {
                ReleasePendingCommandSlot();
            }

            throw;
        }
    }

    /// <inheritdoc />
    public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(CancellationToken cancellationToken)
    {
        // The event channel is SingleReader: only one enumerator may ever drain it, otherwise
        // the two readers would each receive a random subset of events. Claim the reader at CALL
        // time (not lazily on first MoveNext) and fail loudly on a second consumer rather than
        // silently splitting the stream. The distributor pump is the only intended
        // caller; the alarm monitor and dashboard mirror attach to the distributor instead.
        if (Interlocked.CompareExchange(ref _eventsReaderClaimed, 1, 0) != 0)
        {
            throw new InvalidOperationException(
                "WorkerClient.ReadEventsAsync was already claimed by another consumer. The worker event "
                + "channel is single-reader; attach to the SessionEventDistributor instead of draining it twice.");
        }

        return ReadEventsCoreAsync(cancellationToken);
    }

    private async IAsyncEnumerable<WorkerEvent> ReadEventsCoreAsync(
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
        {
            // No metrics call on the hot path: the gauge pulls _eventQueueDepth when scraped.
            Interlocked.Decrement(ref _eventQueueDepth);
            yield return workerEvent;
        }
    }

    /// <inheritdoc />
    public async Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken)
    {
        ThrowIfDisposed();
        if (timeout <= TimeSpan.Zero)
        {
            throw new ArgumentOutOfRangeException(nameof(timeout), timeout, "Shutdown timeout must be greater than zero.");
        }

        WorkerClientState state = State;
        if (state == WorkerClientState.Closed)
        {
            return;
        }

        if (state == WorkerClientState.Faulted)
        {
            KillOwnedProcess("ShutdownFaulted");
            return;
        }

        MarkClosing();
        await EnqueueAsync(CreateShutdownEnvelope(timeout, "gateway-shutdown"), cancellationToken).ConfigureAwait(false);
        _outboundEnvelopes.Writer.TryComplete();

        using CancellationTokenSource timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        timeoutCts.CancelAfter(timeout);
        try
        {
            await WaitForBackgroundTasksAsync(timeoutCts.Token).ConfigureAwait(false);
            await WaitForProcessExitAsync(timeoutCts.Token).ConfigureAwait(false);
            MarkClosed("shutdown");
        }
        catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
        {
            SetFaulted(
                WorkerClientErrorCode.ShutdownTimeout,
                "Worker shutdown timed out.",
                null);
            throw new WorkerClientException(
                WorkerClientErrorCode.ShutdownTimeout,
                $"Worker shutdown timed out after {timeout}.");
        }
    }

    /// <inheritdoc />
    public void Kill(string reason)
    {
        ThrowIfDisposed();
        KillOwnedProcess(reason);
        SetFaulted(
            WorkerClientErrorCode.WorkerFaulted,
            $"Worker was killed by the gateway: {reason}.",
            null);
    }

    /// <summary>Disposes the worker client and releases resources.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync()
    {
        // disposal is the third channel-completing path, and it must enter a terminal state
        // like the other two (MarkClosed, SetFaulted) BEFORE the channels are completed. Otherwise a
        // client disposed while still Ready keeps reporting IsTerminalState() == false, and a read
        // loop already inside DispatchEnvelope (the _stopCts.Cancel below only interrupts the NEXT
        // ReadAsync) takes StageWorkerEvent's overflow arm on the completed staging channel: a
        // spurious ProtocolViolation fault, a worker-event-staging overflow metric and a redundant
        // kill, all during an intentional teardown. MarkClosed is deliberately not reused here: it
        // would overwrite Faulted with Closed and it re-runs the whole teardown sequence this method
        // performs itself. The transition is skipped when the state is already terminal, so a
        // dispose-after-fault keeps Faulted and a double dispose changes nothing.
        bool becameTerminalHere;
        lock (_syncRoot)
        {
            if (_disposed)
            {
                return;
            }

            _disposed = true;
            becameTerminalHere = _state is not (WorkerClientState.Closed or WorkerClientState.Faulted);
            if (becameTerminalHere)
            {
                _state = WorkerClientState.Closed;
            }

            UnregisterEventQueueDepthSource();
        }

        KillOwnedProcess("Dispose");
        _stopCts.Cancel();
        _outboundEnvelopes.Writer.TryComplete();
        _eventStaging.Writer.TryComplete();
        _events.Writer.TryComplete();
        if (becameTerminalHere)
        {
            // A worker that reached Ready and is now being disposed without a prior Close or fault
            // (SessionManager.OpenSessionAsync's failure arm does exactly this) stopped, so the
            // running-worker gauge must come down. RecordWorkerStoppedOnce is latched on
            // _workerStartRecorded, so a client that already recorded its stop through MarkClosed or
            // SetFaulted cannot record a second one here.
            RecordWorkerStoppedOnce("dispose");
        }

        CompletePendingCommands(
            new WorkerClientException(
                WorkerClientErrorCode.GatewayShutdown,
                "Worker client was disposed."));

        await _connection.Stream.DisposeAsync().ConfigureAwait(false);
        using CancellationTokenSource disposeTimeout = new(DisposeTaskTimeout);
        try
        {
            await WaitForBackgroundTasksAsync(disposeTimeout.Token).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            _logger.LogWarning(
                "Timed out waiting for worker client background tasks to stop for session {SessionId}.",
                SessionId);
        }

        _connection.ProcessHandle?.Dispose();
        _pendingCommandSlots.Dispose();
        _stopCts.Dispose();
    }

    /// <summary>
    ///     Worst-case number of bytes <see cref="WriteLoopAsync"/>'s sequence stamp adds to an
    ///     envelope that was sized before it: one tag byte plus a ten-byte varint, the maximum
    ///     encoding of a 64-bit protobuf field. The <c>CommandTooLarge</c> pre-check subtracts this
    ///     from the negotiated frame maximum because it measures the envelope while
    ///     <c>Sequence</c> is still 0 — an unset field costs nothing to encode, so the measured size
    ///     is a lower bound on the frame the writer emits, and without the allowance an envelope in
    ///     the last eleven bytes of the range would pass the pre-check and then fault the session on
    ///     the write.
    /// </summary>
    private const int MaxSequenceEncodingBytes = 11;

    /// <summary>Manages writing envelopes to the worker pipe.</summary>
    private async Task WriteLoopAsync()
    {
        try
        {
            await foreach (WorkerEnvelope envelope in _outboundEnvelopes.Reader.ReadAllAsync(_stopCts.Token).ConfigureAwait(false))
            {
                // stamp the sequence at the point of writing, not when the envelope is built.
                // Stamping at construction let two concurrent InvokeAsync callers take 1 and 2 and then
                // enqueue in the order 2, 1 — non-monotonic on the wire, breaking gateway.md's
                // "monotonic per sender" contract. This loop is the channel's single consumer
                // (SingleReader = true), so wire order and stamp order are the same thing here and
                // _nextSequence needs no interlocking. Mirrors the worker's equivalent fix.
                envelope.Sequence = unchecked(++_nextSequence);
                await _writer.WriteAsync(envelope, _stopCts.Token).ConfigureAwait(false);
            }
        }
        catch (OperationCanceledException) when (_stopCts.IsCancellationRequested || IsTerminalState())
        {
        }
        catch (Exception exception)
        {
            SetFaulted(
                WorkerClientErrorCode.WriteFailed,
                "Worker pipe write failed.",
                exception);
        }
    }

    /// <summary>Manages reading envelopes from the worker pipe.</summary>
    private async Task ReadLoopAsync()
    {
        try
        {
            while (!_stopCts.IsCancellationRequested)
            {
                WorkerEnvelope envelope = await _reader.ReadAsync(_stopCts.Token).ConfigureAwait(false);
                DispatchEnvelope(envelope);
            }
        }
        catch (OperationCanceledException) when (_stopCts.IsCancellationRequested || IsTerminalState())
        {
        }
        catch (WorkerFrameProtocolException exception) when (exception.ErrorCode == WorkerFrameProtocolErrorCode.EndOfStream)
        {
            SetFaulted(
                WorkerClientErrorCode.PipeDisconnected,
                "Worker pipe disconnected.",
                exception);
        }
        catch (Exception exception)
        {
            SetFaulted(
                WorkerClientErrorCode.ProtocolViolation,
                "Worker read loop failed.",
                exception);
        }
    }

    /// <summary>
    ///     Monitors worker heartbeat and detects stale sessions. Mirrors the
    ///     worker-side watchdog: while a command is in flight on the
    ///     gateway↔worker pipe, the heartbeat watchdog is suppressed up to
    ///     <see cref="WorkerClientOptions.HeartbeatStuckCeiling"/> — the worker
    ///     may be busy executing a slow STA command and the heartbeat write may
    ///     be queued behind a long event-drain burst, neither of which
    ///     indicates the worker is actually hung. Once the oldest pending
    ///     command exceeds the ceiling, the fault fires anyway so a truly stuck
    ///     COM call doesn't hide the worker forever.
    /// </summary>
    private async Task HeartbeatLoopAsync()
    {
        try
        {
            while (!_stopCts.IsCancellationRequested)
            {
                await Task.Delay(_options.HeartbeatCheckInterval, _stopCts.Token).ConfigureAwait(false);
                if (State != WorkerClientState.Ready)
                {
                    continue;
                }

                DateTimeOffset lastHeartbeatAt = LastHeartbeatAt;
                DateTimeOffset now = _timeProvider.GetUtcNow();
                if (now - lastHeartbeatAt <= _options.HeartbeatGrace)
                {
                    continue;
                }

                if (TryGetOldestPendingCommandAge(out TimeSpan oldestCommandAge)
                    && oldestCommandAge <= _options.HeartbeatStuckCeiling)
                {
                    continue;
                }

                _metrics?.HeartbeatFailed(SessionId);
                SetFaulted(
                    WorkerClientErrorCode.HeartbeatExpired,
                    $"Worker heartbeat expired. Last heartbeat was at {lastHeartbeatAt:O}.",
                    null);
            }
        }
        catch (OperationCanceledException) when (_stopCts.IsCancellationRequested || IsTerminalState())
        {
        }
    }

    /// <summary>
    ///     Returns the age of the oldest pending command on the worker pipe,
    ///     measured via <see cref="TimeProvider.GetElapsedTime(long)"/> against
    ///     <see cref="PendingCommand.StartTimestamp"/>, or <c>false</c> when no
    ///     commands are pending. Used by the heartbeat watchdog
    ///     to decide whether a heartbeat gap is plausibly the result of
    ///     pipe-write contention rather than a hung worker.
    /// </summary>
    private bool TryGetOldestPendingCommandAge(out TimeSpan oldestAge)
    {
        long oldestStart = long.MaxValue;
        foreach (PendingCommand pending in _pendingCommands.Values)
        {
            if (pending.StartTimestamp < oldestStart)
            {
                oldestStart = pending.StartTimestamp;
            }
        }

        if (oldestStart == long.MaxValue)
        {
            oldestAge = TimeSpan.Zero;
            return false;
        }

        oldestAge = _timeProvider.GetElapsedTime(oldestStart);
        return true;
    }

    /// <summary>
    ///     Routes a received envelope to its handler. Every branch dispatches synchronously and
    ///     immediately — the event branch only stages the event for the dedicated writer — so a full
    ///     event channel can never delay a command reply, heartbeat, fault, or shutdown ack behind an
    ///     event backlog.
    /// </summary>
    /// <param name="envelope">The envelope to dispatch.</param>
    private void DispatchEnvelope(WorkerEnvelope envelope)
    {
        switch (envelope.BodyCase)
        {
            case WorkerEnvelope.BodyOneofCase.WorkerCommandReply:
                CompleteCommand(envelope);
                break;
            case WorkerEnvelope.BodyOneofCase.WorkerEvent:
                StageWorkerEvent(envelope.WorkerEvent);
                break;
            case WorkerEnvelope.BodyOneofCase.WorkerHeartbeat:
                MarkHeartbeat(envelope.WorkerHeartbeat);
                break;
            case WorkerEnvelope.BodyOneofCase.WorkerFault:
                SetFaulted(
                    WorkerClientErrorCode.WorkerFaulted,
                    CreateWorkerFaultMessage(envelope.WorkerFault),
                    null);
                break;
            case WorkerEnvelope.BodyOneofCase.WorkerShutdownAck:
                MarkClosed("worker-shutdown-ack");
                break;
            default:
                SetFaulted(
                    WorkerClientErrorCode.ProtocolViolation,
                    $"Worker sent unexpected envelope body {envelope.BodyCase}.",
                    null);
                break;
        }
    }

    /// <summary>
    ///     Hands a received worker event to the dedicated event writer without blocking the read loop.
    ///     <c>TryWrite</c> is non-blocking, so the read loop keeps dispatching replies, heartbeats and
    ///     faults regardless of how backed up the event path is. It returns <c>false</c> in two cases:
    ///     the staging channel has been completed by one of the three completing paths —
    ///     <see cref="MarkClosed"/>, <see cref="SetFaulted"/> or <see cref="DisposeAsync"/>, each of
    ///     which enters a terminal state before completing the channels, so the event is
    ///     safely dropped because the client is already terminal — or the channel is full at its
    ///     2 × <see cref="WorkerClientOptions.EventChannelCapacity"/> bound. The latter means the
    ///     consumer has been draining slower than the worker produces for the whole time it took the
    ///     worker to emit that many further events, so the client is faulted immediately —
    ///     <see cref="SetFaulted"/> is non-blocking, so the read loop still never awaits here. The
    ///     complementary full-stall case (a consumer that stops entirely) is caught earlier by the
    ///     timed write in <see cref="EnqueueWorkerEventAsync"/>.
    /// </summary>
    /// <param name="workerEvent">The event received from the worker.</param>
    private void StageWorkerEvent(WorkerEvent workerEvent)
    {
        if (workerEvent.Event is not null)
        {
            _metrics?.EventReceived(SessionId, workerEvent.Event.Family.ToString());
        }

        if (_eventStaging.Writer.TryWrite(workerEvent))
        {
            // Counted here rather than at the _events write so the single gauge reports total
            // undelivered events (staged + queued). ReadEventsCoreAsync decrements on consumer read.
            Interlocked.Increment(ref _eventQueueDepth);
            return;
        }

        if (IsTerminalState())
        {
            // Shutdown completed the staging channel; dropping the event is the documented behavior.
            return;
        }

        _metrics?.QueueOverflow("worker-event-staging");
        int depthAtOverflow = Volatile.Read(ref _eventQueueDepth);
        SetFaulted(
            WorkerClientErrorCode.ProtocolViolation,
            $"Worker event staging channel is full at its {_eventStagingCapacity}-event bound "
                + $"(2 x EventChannelCapacity {_options.EventChannelCapacity}); undelivered depth is "
                + $"{depthAtOverflow}. The event consumer is draining slower than the worker produces. "
                + $"Attach or unblock the StreamEvents consumer or raise MxGateway:Events:QueueCapacity.",
            null);
    }

    /// <summary>
    ///     Drains staged worker events and applies the bounded-channel backpressure (and
    ///     sustained-overflow fault) on a dedicated task, so the timed <see cref="Channel"/> write
    ///     never runs on the read loop. Mirrors <see cref="WriteLoopAsync"/> for events.
    /// </summary>
    private async Task EventWriteLoopAsync()
    {
        try
        {
            await foreach (WorkerEvent workerEvent in
                _eventStaging.Reader.ReadAllAsync(_stopCts.Token).ConfigureAwait(false))
            {
                await EnqueueWorkerEventAsync(workerEvent, _stopCts.Token).ConfigureAwait(false);
            }
        }
        catch (OperationCanceledException) when (_stopCts.IsCancellationRequested || IsTerminalState())
        {
        }
    }

    /// <summary>
    ///     Enqueues a worker event for client consumption. The channel is
    ///     configured with <see cref="BoundedChannelFullMode.Wait"/>
    ///     and a brief consumer hiccup is tolerated for up to
    ///     <see cref="WorkerClientOptions.EventChannelFullModeTimeout"/>
    ///     (default 5s) before the worker is faulted. The channel previously
    ///     used <c>TryWrite</c> (non-blocking), which never honored the
    ///     configured <c>FullModeTimeout</c> — the worker faulted on the first
    ///     missed slot even though the wait-mode channel would have absorbed
    ///     the burst. The diagnostic now names the capacity, current depth, and
    ///     the actionable fix (attach <c>StreamEvents</c> or raise
    ///     <c>MxGateway:Events:QueueCapacity</c>). This is the full-stall half of
    ///     the backpressure policy; a consumer that merely drains too slowly is
    ///     caught by the staging bound in <see cref="StageWorkerEvent"/>.
    ///     Queue depth is not counted here — the event was already counted when it
    ///     was staged, so moving it between the two channels changes nothing.
    /// </summary>
    /// <param name="workerEvent">The event to enqueue.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    private async Task EnqueueWorkerEventAsync(
        WorkerEvent workerEvent,
        CancellationToken cancellationToken)
    {
        if (_events.Writer.TryWrite(workerEvent))
        {
            return;
        }

        using CancellationTokenSource fullModeCts = CancellationTokenSource
            .CreateLinkedTokenSource(cancellationToken);
        fullModeCts.CancelAfter(_options.EventChannelFullModeTimeout);
        try
        {
            await _events.Writer.WriteAsync(workerEvent, fullModeCts.Token).ConfigureAwait(false);
            return;
        }
        catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
        {
            // Only the full-mode timeout fired — the outer cancellation is
            // a different concern and is rethrown by the await above when it
            // triggers.
        }

        _metrics?.QueueOverflow("worker-events");
        int depthAtOverflow = Volatile.Read(ref _eventQueueDepth);
        SetFaulted(
            WorkerClientErrorCode.ProtocolViolation,
            $"Worker event channel rejected an event after waiting "
                + $"{_options.EventChannelFullModeTimeout.TotalMilliseconds:F0} ms; "
                + $"undelivered depth is {depthAtOverflow} against a consumer channel "
                + $"of {_options.EventChannelCapacity} capacity. "
                + $"Attach a StreamEvents consumer or raise MxGateway:Events:QueueCapacity.",
            null);
    }

    /// <summary>Completes pending command with worker reply.</summary>
    /// <param name="envelope">Envelope containing command reply.</param>
    private void CompleteCommand(WorkerEnvelope envelope)
    {
        string correlationId = envelope.CorrelationId;
        if (string.IsNullOrWhiteSpace(correlationId))
        {
            correlationId = envelope.WorkerCommandReply.Reply?.CorrelationId ?? string.Empty;
        }

        if (!_pendingCommands.TryRemove(correlationId, out PendingCommand? pendingCommand))
        {
            _logger.LogDebug(
                "Ignoring late or unknown worker command reply for session {SessionId} and correlation {CorrelationId}.",
                SessionId,
                correlationId);
            return;
        }

        ReleasePendingCommandSlot();
        TimeSpan duration = _timeProvider.GetElapsedTime(pendingCommand.StartTimestamp);
        _metrics?.CommandSucceeded(pendingCommand.Method, duration);
        pendingCommand.SetResult(envelope.WorkerCommandReply);
    }

    /// <summary>Fails a pending command with an error.</summary>
    /// <param name="correlationId">Command correlation ID.</param>
    /// <param name="pendingCommand">The pending command.</param>
    /// <param name="errorCode">Error code.</param>
    /// <param name="message">Error message.</param>
    /// <returns>
    ///     <c>true</c> when this call removed the pending entry and owns the failure; <c>false</c> when
    ///     the entry was already gone — a reply, fault, or shutdown got there first, so the caller must
    ///     not take any further action on behalf of that correlation.
    /// </returns>
    private bool RemovePendingCommandAsFailed(
        string correlationId,
        PendingCommand pendingCommand,
        WorkerClientErrorCode errorCode,
        string message)
    {
        if (!_pendingCommands.TryRemove(correlationId, out _))
        {
            return false;
        }

        ReleasePendingCommandSlot();
        TimeSpan duration = _timeProvider.GetElapsedTime(pendingCommand.StartTimestamp);
        _metrics?.CommandFailed(pendingCommand.Method, errorCode.ToString(), duration);
        pendingCommand.SetException(new WorkerClientException(errorCode, message));
        return true;
    }

    /// <summary>
    ///     Fails the pending command for a caller whose own token fired — a client deadline shorter
    ///     than the gateway's command timeout, or a disconnect that cancelled the RPC's token — and
    ///     forwards a <c>WorkerCancel</c> for the correlation the caller has abandoned.
    ///     <para>
    ///     Caller cancellation is the common abandonment path, not the rare one, and it leaves the
    ///     worker in exactly the state the timeout arm's cancel exists for: the correlation is still
    ///     on the worker's single STA queue and would execute in its turn, behind live callers, for
    ///     a caller that is already gone. Under a burst of deadline-cancelled calls that backlog is
    ///     precisely what the cancel was added to shorten, so both arms send it.
    ///     </para>
    ///     <para>
    ///     Gated the same way as the timeout arm: only the caller that owned the removal sends, so a
    ///     reply that won the race is never chased. Additionally skipped once <c>_stopCts</c> has
    ///     fired — the client is closing, the outbound channel is being completed and every pending
    ///     command failed anyway, so a cancel would be at best pointless and at worst a write racing
    ///     teardown.
    ///     </para>
    /// </summary>
    /// <param name="correlationId">Correlation id of the abandoned command.</param>
    /// <param name="pendingCommand">The pending command entry to fail.</param>
    /// <param name="method">Command method name, for the cancel reason and diagnostics.</param>
    private void AbandonCommandOnCallerCancellation(
        string correlationId,
        PendingCommand pendingCommand,
        string method)
    {
        bool removed = RemovePendingCommandAsFailed(
            correlationId,
            pendingCommand,
            WorkerClientErrorCode.CallerCancelled,
            "Command wait was canceled.");

        if (removed && !_stopCts.IsCancellationRequested)
        {
            TrySendCancelForAbandonedCommand(correlationId, method, "caller cancelled the command");
        }
    }

    /// <summary>
    ///     Forwards a <c>WorkerCancel</c> for a correlation the gateway has given up waiting for, so the
    ///     worker can drop it from its STA queue (<c>WorkerPipeSession</c> routes the envelope to
    ///     <c>CancelCommand</c>). A cancel that arrives after the command already reached the COM call
    ///     is a no-op — MXAccess offers no way to abort an in-flight call — so this shortens the STA
    ///     backlog rather than freeing a call already running on it.
    ///     <para>
    ///     A command whose envelope has not yet left <c>_outboundEnvelopes</c> is handled by the same
    ///     path rather than by pulling it back out: <see cref="Channel{T}"/> exposes no removal, and
    ///     the queue is FIFO, so the worker simply reads the command and then its cancel and drops the
    ///     correlation before it ever reaches the STA. Nothing is gained by dequeuing it here.
    ///     </para>
    /// </summary>
    /// <remarks>
    ///     The whole body sits under one catch-all that debug-logs, because the guarantee this method
    ///     owes its caller is structural, not incidental: the caller is on the throw path for the
    ///     timeout or the cancellation, so anything escaping here — an envelope that failed to build,
    ///     a <c>TryWrite</c> against a disposed channel, a scheduler refusing the detached task —
    ///     would replace the <see cref="WorkerClientErrorCode.CommandTimeout"/> (or the
    ///     <see cref="OperationCanceledException"/>) the caller is owed with an unrelated
    ///     exception. Losing the cancel costs the worker one wasted command; losing the timeout
    ///     misreports why the call failed. The detached task carries its own handler for the same
    ///     reason: its failures (including the <see cref="ObjectDisposedException"/> from
    ///     <c>_stopCts</c> if the client is disposed underneath it) happen after this method returns
    ///     and would otherwise be unobserved. It is deliberately not tracked or awaited — it holds no
    ///     resource the shutdown path needs back, and the outbound channel is completed on close.
    /// </remarks>
    /// <param name="correlationId">Correlation id of the abandoned command.</param>
    /// <param name="method">Command method name, for diagnostics.</param>
    /// <param name="reason">Why the gateway abandoned the command, carried on the <c>WorkerCancel</c>.</param>
    private void TrySendCancelForAbandonedCommand(
        string correlationId,
        string method,
        string reason)
    {
        try
        {
            WorkerEnvelope cancelEnvelope = CreateEnvelope(
                correlationId,
                envelope => envelope.WorkerCancel = new WorkerCancel
                {
                    Reason = reason,
                });

            if (_outboundEnvelopes.Writer.TryWrite(cancelEnvelope))
            {
                return;
            }

            _ = Task.Run(async () =>
            {
                try
                {
                    await EnqueueAsync(cancelEnvelope, _stopCts.Token).ConfigureAwait(false);
                }
                catch (Exception exception)
                {
                    LogCancelNotForwarded(exception, method, correlationId);
                }
            });
        }
        catch (Exception exception)
        {
            LogCancelNotForwarded(exception, method, correlationId);
        }
    }

    /// <summary>Records a cancel that could not be forwarded for an abandoned command.</summary>
    /// <param name="exception">The failure that stopped the cancel from being sent.</param>
    /// <param name="method">Command method name of the abandoned command.</param>
    /// <param name="correlationId">Correlation id of the abandoned command.</param>
    private void LogCancelNotForwarded(
        Exception exception,
        string method,
        string correlationId)
    {
        _logger.LogDebug(
            exception,
            "Could not forward a cancel for abandoned worker command {Method} on session {SessionId} "
            + "and correlation {CorrelationId}.",
            method,
            SessionId,
            correlationId);
    }

    /// <summary>Reads and validates a handshake envelope.</summary>
    /// <param name="expectedBody">Expected envelope body type.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>The read envelope.</returns>
    private async Task<WorkerEnvelope> ReadHandshakeEnvelopeAsync(
        WorkerEnvelope.BodyOneofCase expectedBody,
        CancellationToken cancellationToken)
    {
        WorkerEnvelope envelope = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false);

        // a fault frame during the handshake is the worker reporting a structured startup
        // failure — most often MXAccess COM creation (not installed, not registered, access denied)
        // — not a protocol problem. Flattening it to ProtocolViolation discarded the category and
        // diagnostic exactly where they matter most: first session open on a freshly deployed host,
        // where the message reaches the operator log and the client's OpenFailed detail. Use the
        // same formatter the steady-state fault path uses so both windows read alike.
        if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerFault
            && expectedBody != WorkerEnvelope.BodyOneofCase.WorkerFault)
        {
            throw new WorkerClientException(
                WorkerClientErrorCode.WorkerFaulted,
                CreateWorkerFaultMessage(envelope.WorkerFault));
        }

        if (envelope.BodyCase != expectedBody)
        {
            throw new WorkerClientException(
                WorkerClientErrorCode.ProtocolViolation,
                $"Worker handshake expected {expectedBody} but received {envelope.BodyCase}.");
        }

        return envelope;
    }

    /// <summary>Validates worker hello message protocol and nonce.</summary>
    /// <param name="workerHello">The hello message to validate.</param>
    private void ValidateWorkerHello(WorkerHello workerHello)
    {
        if (workerHello.ProtocolVersion != _connection.FrameOptions.ProtocolVersion)
        {
            throw new WorkerClientException(
                WorkerClientErrorCode.ProtocolViolation,
                "Worker hello protocol version does not match the gateway protocol version.");
        }

        if (!string.Equals(workerHello.Nonce, _connection.Nonce, StringComparison.Ordinal))
        {
            throw new WorkerClientException(
                WorkerClientErrorCode.ProtocolViolation,
                "Worker hello nonce does not match the gateway nonce.");
        }

        lock (_syncRoot)
        {
            _processId = workerHello.WorkerProcessId == 0
                ? _connection.ProcessHandle?.ProcessId
                : workerHello.WorkerProcessId;
        }
    }

    /// <summary>Marks the worker as ready and records startup metrics.</summary>
    /// <param name="ready">The ready message.</param>
    private void MarkReady(WorkerReady ready)
    {
        lock (_syncRoot)
        {
            _processId = ready.WorkerProcessId == 0
                ? _processId ?? _connection.ProcessHandle?.ProcessId
                : ready.WorkerProcessId;
            _lastHeartbeatAt = _timeProvider.GetUtcNow();
            _state = WorkerClientState.Ready;
            _workerStartRecorded = true;
        }

        DateTimeOffset readyAt = _timeProvider.GetUtcNow();
        DateTimeOffset launchedAt = _connection.ProcessHandle?.LaunchedAt ?? readyAt;
        _metrics?.WorkerStarted(readyAt - launchedAt);
    }

    /// <summary>Updates the heartbeat timestamp and process ID.</summary>
    /// <param name="heartbeat">The heartbeat message.</param>
    private void MarkHeartbeat(WorkerHeartbeat heartbeat)
    {
        lock (_syncRoot)
        {
            _lastHeartbeatAt = _timeProvider.GetUtcNow();
            if (heartbeat.WorkerProcessId != 0)
            {
                _processId = heartbeat.WorkerProcessId;
            }
        }
    }

    /// <summary>Transitions to closing state.</summary>
    private void MarkClosing()
    {
        lock (_syncRoot)
        {
            if (_state is WorkerClientState.Closed or WorkerClientState.Faulted)
            {
                return;
            }

            _state = WorkerClientState.Closing;
        }
    }

    /// <summary>Marks client as closed and cleans up resources.</summary>
    /// <param name="reason">Reason for closure.</param>
    private void MarkClosed(string reason)
    {
        lock (_syncRoot)
        {
            if (_state == WorkerClientState.Closed)
            {
                return;
            }

            _state = WorkerClientState.Closed;
            UnregisterEventQueueDepthSource();
        }

        _stopCts.Cancel();
        _outboundEnvelopes.Writer.TryComplete();
        _eventStaging.Writer.TryComplete();
        _events.Writer.TryComplete();
        CompletePendingCommands(
            new WorkerClientException(
                WorkerClientErrorCode.GatewayShutdown,
                $"Worker client closed because {reason}."));
        RecordWorkerStoppedOnce(reason);
    }

    /// <summary>Marks client as faulted and propagates the error.</summary>
    /// <param name="errorCode">Error code.</param>
    /// <param name="message">Error message.</param>
    /// <param name="exception">Optional inner exception.</param>
    private void SetFaulted(
        WorkerClientErrorCode errorCode,
        string message,
        Exception? exception)
    {
        WorkerClientException fault = exception is null
            ? new WorkerClientException(errorCode, message)
            : new WorkerClientException(errorCode, message, exception);

        lock (_syncRoot)
        {
            if (_state is WorkerClientState.Faulted or WorkerClientState.Closed)
            {
                return;
            }

            _state = WorkerClientState.Faulted;
            UnregisterEventQueueDepthSource();
        }

        _stopCts.Cancel();
        _outboundEnvelopes.Writer.TryComplete(fault);
        _eventStaging.Writer.TryComplete(fault);
        _events.Writer.TryComplete(fault);
        KillOwnedProcess(errorCode.ToString());
        CompletePendingCommands(fault);
        RecordWorkerStoppedOnce(errorCode.ToString());
        _metrics?.Fault(errorCode.ToString());
        _logger.LogWarning(exception, "Worker client faulted for session {SessionId}: {Message}", SessionId, message);
    }

    /// <summary>
    ///     Removes this terminal client's abandoned backlog from the gateway-wide queue-depth
    ///     gauge. The registration handle is idempotent, so terminal transitions and later disposal
    ///     may all call this safely while the faulted client remains retained for diagnostics.
    /// </summary>
    private void UnregisterEventQueueDepthSource()
    {
        _eventQueueDepthRegistration?.Dispose();
    }

    private void KillOwnedProcess(string reason)
    {
        WorkerProcessHandle? processHandle = _connection.ProcessHandle;
        if (processHandle is null)
        {
            return;
        }

        try
        {
            if (!processHandle.Process.HasExited)
            {
                processHandle.Process.Kill(entireProcessTree: true);
                _metrics?.WorkerKilled(reason);
            }
        }
        catch (Exception exception)
        {
            _logger.LogWarning(
                exception,
                "Failed to kill worker process {ProcessId} for session {SessionId}.",
                processHandle.ProcessId,
                SessionId);
        }
    }

    /// <summary>Records worker stopped metric only once.</summary>
    /// <param name="reason">Reason for stopping.</param>
    private void RecordWorkerStoppedOnce(string reason)
    {
        bool shouldRecord;
        lock (_syncRoot)
        {
            shouldRecord = _workerStartRecorded;
            _workerStartRecorded = false;
        }

        if (shouldRecord)
        {
            _metrics?.WorkerStopped(reason);
        }
    }

    /// <summary>Fails all pending commands with the given exception.</summary>
    /// <param name="exception">Exception to apply to all commands.</param>
    private void CompletePendingCommands(Exception exception)
    {
        foreach (KeyValuePair<string, PendingCommand> item in _pendingCommands.ToArray())
        {
            if (_pendingCommands.TryRemove(item.Key, out PendingCommand? pendingCommand))
            {
                ReleasePendingCommandSlot();
                TimeSpan duration = _timeProvider.GetElapsedTime(pendingCommand.StartTimestamp);
                _metrics?.CommandFailed(pendingCommand.Method, exception.GetType().Name, duration);
                pendingCommand.SetException(exception);
            }
        }
    }

    /// <summary>Releases a pending command slot.</summary>
    private void ReleasePendingCommandSlot()
    {
        try
        {
            _pendingCommandSlots.Release();
        }
        catch (SemaphoreFullException)
        {
        }
    }

    /// <summary>Transitions from created to handshaking state.</summary>
    private void TransitionFromCreatedToHandshaking()
    {
        lock (_syncRoot)
        {
            if (_state != WorkerClientState.Created)
            {
                throw new WorkerClientException(
                    WorkerClientErrorCode.InvalidState,
                    $"Worker client cannot start from state {_state}.");
            }

            _state = WorkerClientState.Handshaking;
        }
    }

    /// <summary>Throws if client is not in ready state.</summary>
    private void EnsureReady()
    {
        WorkerClientState state = State;
        if (state != WorkerClientState.Ready)
        {
            throw new WorkerClientException(
                WorkerClientErrorCode.InvalidState,
                $"Worker client is not ready. Current state is {state}.");
        }
    }

    /// <summary>Checks if current state is terminal.</summary>
    /// <returns>True if closed, closing, or faulted.</returns>
    private bool IsTerminalState()
    {
        WorkerClientState state = State;
        return state is WorkerClientState.Closing or WorkerClientState.Closed or WorkerClientState.Faulted;
    }

    /// <summary>Enqueues an envelope for writing to the worker.</summary>
    /// <param name="envelope">Envelope to enqueue.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    private async Task EnqueueAsync(
        WorkerEnvelope envelope,
        CancellationToken cancellationToken)
    {
        try
        {
            await _outboundEnvelopes.Writer.WriteAsync(envelope, cancellationToken).ConfigureAwait(false);
        }
        catch (ChannelClosedException exception)
        {
            throw new WorkerClientException(
                WorkerClientErrorCode.WriteFailed,
                "Worker outbound channel is closed.",
                exception);
        }
    }

    /// <summary>Creates gateway hello envelope.</summary>
    /// <returns>The hello envelope.</returns>
    private WorkerEnvelope CreateGatewayHelloEnvelope()
    {
        return CreateEnvelope(
            correlationId: string.Empty,
            envelope => envelope.GatewayHello = new GatewayHello
            {
                SupportedProtocolVersion = _connection.FrameOptions.ProtocolVersion,
                Nonce = _connection.Nonce,
                GatewayVersion = typeof(GatewayContractInfo).Assembly.GetName().Version?.ToString() ?? GatewayVersionFallback,

                // Convey the negotiated worker-frame maximum so the worker adopts it instead of a
                // hard-coded default. Sits above the public gRPC cap by the envelope reserve.
                MaxFrameBytes = (uint)_connection.FrameOptions.MaxMessageBytes,
            });
    }

    /// <summary>Creates command envelope.</summary>
    /// <param name="correlationId">Command correlation ID.</param>
    /// <param name="command">The command to wrap.</param>
    /// <returns>The command envelope.</returns>
    private WorkerEnvelope CreateCommandEnvelope(
        string correlationId,
        WorkerCommand command)
    {
        // no second clone. MxAccessGrpcMapper.MapCommand already deep-cloned the command
        // out of the caller-owned gRPC MxCommandRequest, so this WorkerCommand is a fresh graph
        // owned by the invoke pipeline and referenced by no other consumer. The envelope is built
        // and owned entirely inside WorkerClient and the command is not touched again after this
        // point, so transferring it into the envelope introduces no aliasing hazard.
        return CreateEnvelope(
            correlationId,
            envelope => envelope.WorkerCommand = command);
    }

    /// <summary>Creates shutdown envelope.</summary>
    /// <param name="timeout">Shutdown timeout.</param>
    /// <param name="reason">Shutdown reason.</param>
    /// <returns>The shutdown envelope.</returns>
    private WorkerEnvelope CreateShutdownEnvelope(
        TimeSpan timeout,
        string reason)
    {
        return CreateEnvelope(
            correlationId: string.Empty,
            envelope => envelope.WorkerShutdown = new WorkerShutdown
            {
                GracePeriod = Duration.FromTimeSpan(timeout),
                Reason = reason,
            });
    }

    /// <summary>Creates a new worker envelope with common fields.</summary>
    /// <param name="correlationId">Correlation ID for the envelope.</param>
    /// <param name="setBody">Action to set envelope body.</param>
    /// <returns>The created envelope.</returns>
    private WorkerEnvelope CreateEnvelope(
        string correlationId,
        Action<WorkerEnvelope> setBody)
    {
        // Sequence is deliberately left unset here: WriteLoopAsync stamps it immediately before the
        // frame goes out, so the numbers are monotonic in wire order however the callers interleave
        // between construction and enqueue (mirroring the worker's fix).
        WorkerEnvelope envelope = new()
        {
            ProtocolVersion = _connection.FrameOptions.ProtocolVersion,
            SessionId = SessionId,
            CorrelationId = correlationId,
        };
        setBody(envelope);

        return envelope;
    }

    /// <summary>Gets the human-readable command method name.</summary>
    /// <param name="command">The command to get method name from.</param>
    /// <returns>Command method name.</returns>
    private static string GetCommandMethod(WorkerCommand command)
    {
        return command.Command?.Kind.ToString() ?? MxCommandKind.Unspecified.ToString();
    }

    /// <summary>Creates a fault message from worker fault data.</summary>
    /// <param name="fault">The worker fault.</param>
    /// <returns>Formatted fault message.</returns>
    private static string CreateWorkerFaultMessage(WorkerFault fault)
    {
        return string.IsNullOrWhiteSpace(fault.DiagnosticMessage)
            ? $"Worker faulted with category {fault.Category}."
            : $"Worker faulted with category {fault.Category}: {fault.DiagnosticMessage}";
    }

    /// <summary>Waits for all background tasks to complete.</summary>
    /// <param name="cancellationToken">Cancellation token.</param>
    private async Task WaitForBackgroundTasksAsync(CancellationToken cancellationToken)
    {
        Task[] tasks = new[] { _readLoopTask, _writeLoopTask, _eventWriteLoopTask, _heartbeatLoopTask }
            .Where(task => task is not null)
            .Cast<Task>()
            .ToArray();

        if (tasks.Length == 0)
        {
            return;
        }

        await Task.WhenAll(tasks).WaitAsync(cancellationToken).ConfigureAwait(false);
    }

    /// <summary>Waits for the worker process to exit.</summary>
    /// <param name="cancellationToken">Cancellation token.</param>
    private async Task WaitForProcessExitAsync(CancellationToken cancellationToken)
    {
        WorkerProcessHandle? processHandle = _connection.ProcessHandle;
        if (processHandle is null || processHandle.Process.HasExited)
        {
            return;
        }

        await processHandle.Process.WaitForExitAsync(cancellationToken).ConfigureAwait(false);
    }

    /// <summary>Throws if the client has been disposed.</summary>
    private void ThrowIfDisposed()
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
    }

    private sealed class PendingCommand
    {
        private readonly TaskCompletionSource<WorkerCommandReply> _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);

        /// <summary>Initializes a pending command awaiting a worker reply.</summary>
        /// <param name="correlationId">Command correlation ID for reply matching.</param>
        /// <param name="method">Command method name.</param>
        /// <param name="startTimestamp">Start time in milliseconds for duration tracking.</param>
        public PendingCommand(
            string correlationId,
            string method,
            long startTimestamp)
        {
            CorrelationId = correlationId;
            Method = method;
            StartTimestamp = startTimestamp;
        }

        /// <summary>Gets the command correlation ID.</summary>
        public string CorrelationId { get; }

        /// <summary>Gets the command method name.</summary>
        public string Method { get; }

        /// <summary>Gets the command start timestamp.</summary>
        public long StartTimestamp { get; }

        /// <summary>Gets the task that completes when reply arrives.</summary>
        public Task<WorkerCommandReply> Task => _completion.Task;

        /// <summary>Completes the command with a reply.</summary>
        /// <param name="reply">The command reply.</param>
        public void SetResult(WorkerCommandReply reply)
        {
            _completion.TrySetResult(reply);
        }

        /// <summary>Completes the command with an exception.</summary>
        /// <param name="exception">The exception.</param>
        public void SetException(Exception exception)
        {
            _completion.TrySetException(exception);
        }
    }
}