mxaccessgw/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs

using System.Diagnostics.CodeAnalysis;
using System.Threading.Channels;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Alarms;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Grpc;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;

namespace ZB.MOM.WW.MxGateway.Tests.Alarms;

/// <summary>
///     GWC-26 regression tests for the alarm monitor's startup ordering. Unlike the
///     sibling alarm-monitor tests, the session manager here hands the monitor a REAL
///     <see cref="GatewaySession"/> that is driven to Ready with a dashboard mirror, so the
///     distributor pump is already running when the monitor attaches — the production
///     condition under which a late internal subscriber silently misses everything the pump
///     already fanned.
/// </summary>
public sealed class GatewayAlarmMonitorAttachOrderTests
{
    private const string AlarmReference = "Galaxy!Area.Tank01.Hi";
    private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(30);

    /// <summary>
    ///     The monitor must take its internal distributor lease BEFORE issuing
    ///     <c>SubscribeAlarms</c>, so transitions the worker emits during the
    ///     subscribe + first-reconcile window buffer in the lease instead of being fanned to a
    ///     subscriber set the monitor has not joined yet. The test parks the monitor inside its
    ///     <c>SubscribeAlarms</c> round trip, emits a Raise and an Acknowledge, waits until the
    ///     dashboard mirror proves the pump has already fanned both, and only then releases the
    ///     monitor: with a late attach both transitions are lost to the alarm feed forever.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task TransitionsDuringSubscribeWindow_StillReachTheAlarmFeed()
    {
        using GatewayMetrics metrics = new();
        await using FakeSessionManager sessions = new();
        sessions.HoldSubscribeUntilReleased();
        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);

        using CancellationTokenSource cts = new();
        await monitor.StartAsync(cts.Token);
        await sessions.WaitForSubscribeStartAsync(WaitTimeout);

        // A live feed subscriber, drained past its baseline ProviderStatus so it is registered
        // before any window transition is broadcast.
        List<AlarmFeedMessage> received = [];
        TaskCompletionSource baselineReceived = new(TaskCreationOptions.RunContinuationsAsynchronously);
        using CancellationTokenSource streamCts = new();
        Task reader = ReadFeedAsync(monitor, received, baselineReceived, streamCts.Token);
        await baselineReceived.Task.WaitAsync(WaitTimeout);

        // The window: the worker reports a raise and an acknowledge while the monitor is still
        // waiting for its SubscribeAlarms reply.
        sessions.EmitEvent(Transition(1, AlarmTransitionKind.Raise));
        sessions.EmitEvent(Transition(2, AlarmTransitionKind.Acknowledge));

        // The dashboard mirror is an independent distributor subscriber: once it has both
        // events, the pump has provably already fanned them, so a subscriber that registers
        // after this point can never receive them.
        await WaitUntilAsync(() => sessions.Broadcaster.Captures.Count == 2, WaitTimeout);

        sessions.ReleaseSubscribe();

        AlarmFeedMessage raise = await WaitForAsync(
            received,
            m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
                && m.Transition.TransitionKind == AlarmTransitionKind.Raise,
            WaitTimeout);
        AlarmFeedMessage acknowledge = await WaitForAsync(
            received,
            m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
                && m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge,
            WaitTimeout);

        Assert.Equal(AlarmReference, raise.Transition.AlarmFullReference);
        Assert.Equal(AlarmReference, acknowledge.Transition.AlarmFullReference);

        await streamCts.CancelAsync();
        await reader;
        await cts.CancelAsync();
        await monitor.StopAsync(CancellationToken.None);
    }

    /// <summary>
    ///     Defense in depth for any window the attach reorder cannot cover (worker restart,
    ///     internal-subscriber overflow disconnect): when a reconcile snapshot reports an alarm
    ///     the cache already holds but with the state advanced to
    ///     <see cref="AlarmConditionState.ActiveAcked"/>, the monitor broadcasts an
    ///     <see cref="AlarmTransitionKind.Acknowledge"/> feed transition. Before the fix the
    ///     acked state was absorbed silently by the snapshot replace, leaving live subscribers
    ///     showing the alarm unacked until it cleared.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ApplyReconcileBroadcastsAcknowledgeDelta()
    {
        using GatewayMetrics metrics = new();
        await using FakeSessionManager sessions = new();
        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);

        using CancellationTokenSource cts = new();
        await monitor.StartAsync(cts.Token);
        await sessions.WaitForSubscribeStartAsync(WaitTimeout);

        // Seed the cache with an active (unacked) alarm through a reconcile rather than a live
        // transition: a buffered live transition could still be in flight when the cache first
        // shows the alarm, and would then broadcast to the reader below and pollute the
        // exactly-one-transition assertion. A provider-mode event forces the reconcile
        // immediately, so the test never waits on the periodic timer.
        sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.Active));
        sessions.EmitEvent(ProviderModeProbe(1));
        await WaitUntilAsync(
            () => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference
                && alarm.CurrentState == AlarmConditionState.Active),
            WaitTimeout);

        // Subscribe AFTER the seed: this reader's snapshot carries the alarm, so every
        // transition it observes from here on is a reconcile-derived broadcast.
        List<AlarmFeedMessage> received = [];
        TaskCompletionSource snapshotComplete = new(TaskCreationOptions.RunContinuationsAsynchronously);
        using CancellationTokenSource streamCts = new();
        Task reader = ReadFeedAsync(monitor, received, snapshotComplete, streamCts.Token, untilSnapshotComplete: true);
        await snapshotComplete.Task.WaitAsync(WaitTimeout);

        // The worker now reports the same alarm acknowledged. A provider-mode event forces an
        // immediate reconcile pass so the test does not wait on the periodic timer.
        sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.ActiveAcked));
        sessions.EmitEvent(ProviderModeProbe(2));

        AlarmFeedMessage acknowledge = await WaitForAsync(
            received,
            m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
                && m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge,
            WaitTimeout);
        Assert.Equal(AlarmReference, acknowledge.Transition.AlarmFullReference);

        lock (received)
        {
            AlarmFeedMessage[] transitions = received
                .Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition)
                .ToArray();
            AlarmFeedMessage single = Assert.Single(transitions);
            Assert.Equal(AlarmTransitionKind.Acknowledge, single.Transition.TransitionKind);
        }

        await streamCts.CancelAsync();
        await reader;
        await cts.CancelAsync();
        await monitor.StopAsync(CancellationToken.None);
    }

    /// <summary>
    ///     NEXT-03. A reconcile Raise repair applied while the matching live Raise is still
    ///     buffered must not double-broadcast: the live transition carrying the same worker
    ///     timestamp and resulting state the cache already holds is a duplicate and is suppressed.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task LiveTransitionMatchingReconcileRepair_IsSuppressed()
    {
        using GatewayMetrics metrics = new();
        await using FakeSessionManager sessions = new();
        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);

        using CancellationTokenSource cts = new();
        await monitor.StartAsync(cts.Token);
        await sessions.WaitForSubscribeStartAsync(WaitTimeout);

        // The reconcile sees the raised alarm first (its Raise repair broadcasts before this
        // reader attaches) and stamps the cache with the worker's transition timestamp.
        Timestamp raiseTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero));
        sessions.SetReconcileSnapshot(SnapshotAt(AlarmConditionState.Active, raiseTime, raiseTime));
        sessions.EmitEvent(ProviderModeProbe(1));
        await WaitUntilAsync(
            () => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference),
            WaitTimeout);

        List<AlarmFeedMessage> received = [];
        TaskCompletionSource snapshotComplete = new(TaskCreationOptions.RunContinuationsAsynchronously);
        using CancellationTokenSource streamCts = new();
        Task reader = ReadFeedAsync(monitor, received, snapshotComplete, streamCts.Token, untilSnapshotComplete: true);
        await snapshotComplete.Task.WaitAsync(WaitTimeout);

        // The buffered live Raise drains with the same worker timestamp — a duplicate of the
        // repair. The follow-up Acknowledge with a newer timestamp is genuine and must pass.
        sessions.EmitEvent(TransitionAt(2, AlarmTransitionKind.Raise, raiseTime, raiseTime));
        Timestamp ackTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 12, 0, 5, TimeSpan.Zero));
        sessions.EmitEvent(TransitionAt(3, AlarmTransitionKind.Acknowledge, ackTime, raiseTime));

        await WaitForAsync(
            received,
            m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
                && m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge,
            WaitTimeout);

        lock (received)
        {
            AlarmFeedMessage[] transitions = received
                .Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition)
                .ToArray();
            AlarmFeedMessage single = Assert.Single(transitions);
            Assert.Equal(AlarmTransitionKind.Acknowledge, single.Transition.TransitionKind);
        }

        await streamCts.CancelAsync();
        await reader;
        await cts.CancelAsync();
        await monitor.StopAsync(CancellationToken.None);
    }

    /// <summary>
    ///     NEXT-03, the Clear leg. A reconcile Clear repair removes the cache entry, so the
    ///     buffered live Clear is deduped through the tombstone keyed on the instance's original
    ///     raise timestamp — and a genuinely new raise/clear cycle is never swallowed.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task LiveClearMatchingReconcileClearRepair_IsSuppressed()
    {
        using GatewayMetrics metrics = new();
        await using FakeSessionManager sessions = new();
        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);

        using CancellationTokenSource cts = new();
        await monitor.StartAsync(cts.Token);
        await sessions.WaitForSubscribeStartAsync(WaitTimeout);

        Timestamp raiseTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 13, 0, 0, TimeSpan.Zero));
        sessions.SetReconcileSnapshot(SnapshotAt(AlarmConditionState.Active, raiseTime, raiseTime));
        sessions.EmitEvent(ProviderModeProbe(1));
        await WaitUntilAsync(
            () => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference),
            WaitTimeout);

        List<AlarmFeedMessage> received = [];
        TaskCompletionSource snapshotComplete = new(TaskCreationOptions.RunContinuationsAsynchronously);
        using CancellationTokenSource streamCts = new();
        Task reader = ReadFeedAsync(monitor, received, snapshotComplete, streamCts.Token, untilSnapshotComplete: true);
        await snapshotComplete.Task.WaitAsync(WaitTimeout);

        // The worker no longer reports the alarm: the reconcile synthesizes the Clear repair and
        // tombstones the instance by its original raise timestamp.
        sessions.SetReconcileSnapshot();
        sessions.EmitEvent(ProviderModeProbe(2));
        await WaitForAsync(
            received,
            m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
                && m.Transition.TransitionKind == AlarmTransitionKind.Clear,
            WaitTimeout);

        // The buffered live Clear for the SAME instance is a duplicate of the repair; the Raise
        // that follows starts a new instance and must pass.
        Timestamp clearTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 13, 0, 10, TimeSpan.Zero));
        sessions.EmitEvent(TransitionAt(3, AlarmTransitionKind.Clear, clearTime, raiseTime));
        Timestamp newRaiseTime = Timestamp.FromDateTimeOffset(new DateTimeOffset(2026, 8, 10, 13, 0, 20, TimeSpan.Zero));
        sessions.EmitEvent(TransitionAt(4, AlarmTransitionKind.Raise, newRaiseTime, newRaiseTime));

        await WaitForAsync(
            received,
            m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
                && m.Transition.TransitionKind == AlarmTransitionKind.Raise,
            WaitTimeout);

        lock (received)
        {
            AlarmTransitionKind[] kinds = received
                .Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition)
                .Select(m => m.Transition.TransitionKind)
                .ToArray();
            Assert.Equal([AlarmTransitionKind.Clear, AlarmTransitionKind.Raise], kinds);
        }

        await streamCts.CancelAsync();
        await reader;
        await cts.CancelAsync();
        await monitor.StopAsync(CancellationToken.None);
    }

    /// <summary>
    ///     <see cref="GatewayAlarmMonitor.CurrentAlarms"/> clones the whole active-alarm set under
    ///     the broadcast lock, so rebuilding it per read stalls every transition and broadcast
    ///     behind the copy once the dashboard polls a large alarm set. The projection is memoized
    ///     for as long as the set is unchanged, and every mutation must invalidate it — a stale
    ///     projection would hide live transitions from the dashboard and the QueryActiveAlarms RPC.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task CurrentAlarmsProjectionIsMemoizedUntilTheAlarmSetChanges()
    {
        using GatewayMetrics metrics = new();
        await using FakeSessionManager sessions = new();
        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);

        using CancellationTokenSource cts = new();
        await monitor.StartAsync(cts.Token);
        await sessions.WaitForSubscribeStartAsync(WaitTimeout);

        // Seed through a reconcile (forced by a provider-mode probe) so the cache holds one
        // unacked alarm and no further mutation is in flight.
        sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.Active));
        sessions.EmitEvent(ProviderModeProbe(1));
        await WaitUntilAsync(
            () => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference
                && alarm.CurrentState == AlarmConditionState.Active),
            WaitTimeout);

        IReadOnlyList<ActiveAlarmSnapshot> first = monitor.CurrentAlarms;
        Assert.Same(first, monitor.CurrentAlarms);

        // A live Acknowledge replaces the cached snapshot, so the next read must rebuild.
        sessions.EmitEvent(Transition(2, AlarmTransitionKind.Acknowledge));
        await WaitUntilAsync(
            () => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference
                && alarm.CurrentState == AlarmConditionState.ActiveAcked),
            WaitTimeout);

        IReadOnlyList<ActiveAlarmSnapshot> second = monitor.CurrentAlarms;
        Assert.NotSame(first, second);
        Assert.Same(second, monitor.CurrentAlarms);

        // The pre-transition projection is a snapshot of the old generation, not a live view.
        Assert.Equal(AlarmConditionState.Active, Assert.Single(first).CurrentState);

        await cts.CancelAsync();
        await monitor.StopAsync(CancellationToken.None);
    }

    private static GatewayAlarmMonitor CreateMonitor(FakeSessionManager sessions, GatewayMetrics metrics)
    {
        AlarmsOptions options = new()
        {
            Enabled = true,
            SubscriptionExpression = @"\\NODE\Galaxy!Area",
        };
        return new GatewayAlarmMonitor(
            sessions,
            new StubWatchListResolver(),
            metrics,
            Microsoft.Extensions.Options.Options.Create(new GatewayOptions { Alarms = options }),
            NullLogger<GatewayAlarmMonitor>.Instance);
    }

    // Drains the monitor's feed into received, signalling gate on the first message (the
    // baseline ProviderStatus) or, when untilSnapshotComplete is set, on SnapshotComplete.
    private static Task ReadFeedAsync(
        GatewayAlarmMonitor monitor,
        List<AlarmFeedMessage> received,
        TaskCompletionSource gate,
        CancellationToken cancellationToken,
        bool untilSnapshotComplete = false)
    {
        return Task.Run(
            async () =>
            {
                try
                {
                    await foreach (AlarmFeedMessage message in monitor.StreamAsync(null, cancellationToken))
                    {
                        bool opensGate = !untilSnapshotComplete
                            || message.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotComplete;

                        // Record only what arrives AFTER the gate opened: everything up to and
                        // including the gate message is this subscriber's snapshot preamble, not
                        // a live broadcast, so assertions stay about broadcasts alone.
                        lock (received)
                        {
                            if (gate.Task.IsCompleted)
                            {
                                received.Add(message);
                            }
                        }

                        if (opensGate)
                        {
                            gate.TrySetResult();
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    // Expected when the test cancels the stream.
                }
            },
            CancellationToken.None);
    }

    private static MxEvent Transition(ulong sequence, AlarmTransitionKind kind) => new()
    {
        Family = MxEventFamily.OnAlarmTransition,
        WorkerSequence = sequence,
        OnAlarmTransition = new OnAlarmTransitionEvent
        {
            AlarmFullReference = AlarmReference,
            SourceObjectReference = "Tank01",
            AlarmTypeName = "AnalogLimitAlarm.Hi",
            TransitionKind = kind,
            Severity = 500,
            SourceProvider = AlarmProviderMode.Alarmmgr,
            TransitionTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
        },
    };

    // A no-op provider-mode event. The monitor forces an immediate reconcile after every one,
    // which is how these tests drive a reconcile pass without waiting on the periodic timer.
    private static MxEvent ProviderModeProbe(ulong sequence) => new()
    {
        Family = MxEventFamily.OnAlarmProviderModeChanged,
        WorkerSequence = sequence,
        OnAlarmProviderModeChanged = new OnAlarmProviderModeChangedEvent
        {
            Mode = AlarmProviderMode.Alarmmgr,
            Reason = "probe",
            At = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
        },
    };

    private static ActiveAlarmSnapshot Snapshot(AlarmConditionState state) => new()
    {
        AlarmFullReference = AlarmReference,
        SourceObjectReference = "Tank01",
        AlarmTypeName = "AnalogLimitAlarm.Hi",
        CurrentState = state,
        Severity = 500,
        SourceProvider = AlarmProviderMode.Alarmmgr,
    };

    // Snapshot carrying the worker-side identity markers the NEXT-03 dedup compares on.
    private static ActiveAlarmSnapshot SnapshotAt(
        AlarmConditionState state,
        Timestamp lastTransition,
        Timestamp originalRaise)
    {
        ActiveAlarmSnapshot snapshot = Snapshot(state);
        snapshot.LastTransitionTimestamp = lastTransition;
        snapshot.OriginalRaiseTimestamp = originalRaise;
        return snapshot;
    }

    // Live transition with explicit worker timestamps, for driving the NEXT-03 dedup.
    private static MxEvent TransitionAt(
        ulong sequence,
        AlarmTransitionKind kind,
        Timestamp transitionTimestamp,
        Timestamp originalRaise)
    {
        MxEvent mxEvent = Transition(sequence, kind);
        mxEvent.OnAlarmTransition.TransitionTimestamp = transitionTimestamp;
        mxEvent.OnAlarmTransition.OriginalRaiseTimestamp = originalRaise;
        return mxEvent;
    }

    private static async Task<AlarmFeedMessage> WaitForAsync(
        List<AlarmFeedMessage> received,
        Func<AlarmFeedMessage, bool> predicate,
        TimeSpan timeout)
    {
        DateTime deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            lock (received)
            {
                AlarmFeedMessage? match = received.FirstOrDefault(predicate);
                if (match is not null)
                {
                    return match;
                }
            }

            await Task.Delay(25);
        }

        throw new TimeoutException("No matching AlarmFeedMessage was received in time.");
    }

    private static async Task WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
    {
        DateTime deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (condition())
            {
                return;
            }

            await Task.Delay(25);
        }

        throw new TimeoutException("Condition was not met in time.");
    }

    /// <summary><see cref="IAlarmWatchListResolver"/> that resolves an empty watch-list.</summary>
    private sealed class StubWatchListResolver : IAlarmWatchListResolver
    {
        /// <inheritdoc />
        public Task<AlarmWatchListResolution> ResolveAsync(
            AlarmsOptions options,
            CancellationToken cancellationToken = default) =>
            Task.FromResult(new AlarmWatchListResolution(
                [],
                DiscoveryAttempted: false,
                DiscoveryFailed: false,
                DiscoveryFailureMessage: null));
    }

    /// <summary>
    ///     Session manager that hands the monitor a real <see cref="GatewaySession"/> driven to
    ///     Ready with a dashboard mirror, so the distributor pump is running before the monitor
    ///     attaches. <see cref="EmitEvent"/> pushes worker events through the fake worker client
    ///     into that pump, exactly as a live worker would.
    /// </summary>
    private sealed class FakeSessionManager : ISessionManager, IAsyncDisposable
    {
        private readonly Channel<WorkerEvent> _events = Channel.CreateUnbounded<WorkerEvent>();
        private readonly TaskCompletionSource _subscribeStarted =
            new(TaskCreationOptions.RunContinuationsAsynchronously);
        private readonly object _sync = new();
        private TaskCompletionSource _subscribeGate = CreateReleasedGate();
        private ActiveAlarmSnapshot[] _reconcileSnapshot = [];
        private GatewaySession? _session;

        /// <summary>Dashboard mirror attached to the session; proves what the pump has fanned.</summary>
        public RecordingDashboardEventBroadcaster Broadcaster { get; } = new();

        /// <summary>Re-arms the gate so <c>SubscribeAlarms</c> parks until <see cref="ReleaseSubscribe"/>.</summary>
        public void HoldSubscribeUntilReleased() =>
            _subscribeGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);

        /// <summary>Releases a gate armed by <see cref="HoldSubscribeUntilReleased"/>.</summary>
        public void ReleaseSubscribe() => _subscribeGate.TrySetResult();

        /// <summary>Completes once the monitor's <c>SubscribeAlarms</c> command has arrived.</summary>
        /// <param name="timeout">The maximum time to wait.</param>
        /// <returns>A task that completes when the command arrives.</returns>
        public Task WaitForSubscribeStartAsync(TimeSpan timeout) => _subscribeStarted.Task.WaitAsync(timeout);

        /// <summary>Sets the active-alarm snapshot every <c>QueryActiveAlarms</c> reconcile returns.</summary>
        /// <param name="snapshots">The snapshots to report.</param>
        public void SetReconcileSnapshot(params ActiveAlarmSnapshot[] snapshots)
        {
            lock (_sync)
            {
                _reconcileSnapshot = snapshots;
            }
        }

        /// <summary>Pushes a worker event into the session's distributor pump.</summary>
        /// <param name="mxEvent">The event to push.</param>
        public void EmitEvent(MxEvent mxEvent) =>
            _events.Writer.TryWrite(new WorkerEvent { Event = mxEvent });

        /// <inheritdoc />
        public Task<GatewaySession> OpenSessionAsync(
            SessionOpenRequest request,
            string? clientIdentity,
            string? ownerKeyId,
            CancellationToken cancellationToken)
        {
            GatewaySession session = new(
                sessionId: "session-alarm-attach-order",
                backendName: "Galaxy",
                pipeName: "mxaccess-gateway-1-session-alarm-attach-order",
                nonce: "nonce",
                clientIdentity: clientIdentity,
                ownerKeyId: ownerKeyId,
                clientSessionName: request.ClientSessionName,
                clientCorrelationId: request.ClientCorrelationId,
                commandTimeout: TimeSpan.FromSeconds(30),
                startupTimeout: TimeSpan.FromSeconds(30),
                shutdownTimeout: TimeSpan.FromSeconds(30),
                leaseDuration: TimeSpan.FromMinutes(30),
                openedAt: DateTimeOffset.UtcNow,
                eventStreaming: new SessionEventStreaming(
                    new MxAccessGrpcMapper(),
                    new EventOptions { QueueCapacity = 64 },
                    NullLogger<SessionEventDistributor>.Instance,
                    TimeProvider.System,
                    new GatewayMetrics(),
                    Broadcaster));
            session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader));

            // MarkReady starts the dashboard mirror, and with it the distributor pump — the
            // production precondition this regression depends on.
            session.MarkReady();
            _session = session;
            return Task.FromResult(session);
        }

        /// <inheritdoc />
        public async Task<WorkerCommandReply> InvokeAsync(
            string sessionId,
            WorkerCommand command,
            CancellationToken cancellationToken)
        {
            MxCommandReply reply = new()
            {
                ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
            };

            switch (command.Command?.Kind)
            {
                case MxCommandKind.SubscribeAlarms:
                    _subscribeStarted.TrySetResult();
                    await _subscribeGate.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
                    break;
                case MxCommandKind.QueryActiveAlarms:
                    QueryActiveAlarmsReplyPayload payload = new();
                    lock (_sync)
                    {
                        payload.Snapshots.AddRange(_reconcileSnapshot.Select(snapshot => snapshot.Clone()));
                    }

                    reply.QueryActiveAlarms = payload;
                    break;
            }

            return new WorkerCommandReply { Reply = reply };
        }

        /// <inheritdoc />
        public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
        {
            session = _session;
            return session is not null;
        }

        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(string sessionId, CancellationToken cancellationToken)
        {
            _events.Writer.TryComplete();
            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
        }

        /// <inheritdoc />
        public Task<SessionCloseResult> KillWorkerAsync(string sessionId, string reason, CancellationToken cancellationToken) =>
            Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));

        /// <inheritdoc />
        public Task<int> CloseExpiredLeasesAsync(DateTimeOffset now, CancellationToken cancellationToken) =>
            Task.FromResult(0);

        /// <inheritdoc />
        public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;

        /// <summary>Disposes the session the fake handed out.</summary>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public async ValueTask DisposeAsync()
        {
            _events.Writer.TryComplete();
            if (_session is not null)
            {
                await _session.DisposeAsync().ConfigureAwait(false);
            }
        }

        private static TaskCompletionSource CreateReleasedGate()
        {
            TaskCompletionSource gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
            gate.SetResult();
            return gate;
        }
    }
}