using System.Collections.Concurrent;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;

namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;

/// <summary>
/// Tests for <see cref="SiteAlarmAggregatorActor"/> (plan #10, Task 4) — the transient,
/// in-memory, per-site live alarm aggregator. Covers the tricky bits: seed-then-stream
/// ordering (a delta arriving before the seed completes is neither lost nor double-applied),
/// dedup by AlarmKey, placeholder seeded-but-not-on-stream coherence, NodeA↔NodeB reconnect
/// re-seed, and periodic reconcile authoritative-replace (drift correction).
/// </summary>
public class SiteAlarmAggregatorActorTests : TestKit
{
    private const string SiteId = "site-alpha";
    private const string GrpcNodeA = "http://localhost:5100";
    private const string GrpcNodeB = "http://localhost:5200";
    private const string Instance = "Site1.Pump01";

    public SiteAlarmAggregatorActorTests() : base(@"akka.loglevel = WARNING")
    {
    }

    // Former process-global static test seams, now ctor-injected per actor (R2 T12).
    private static readonly TimeSpan TestReconnectDelay = TimeSpan.FromMilliseconds(50);
    private static readonly TimeSpan TestStabilityWindow = TimeSpan.FromSeconds(30);

    // ── Test doubles ────────────────────────────────────────────────────────────

    /// <summary>
    /// Controllable seed fan-out: each invocation returns a fresh Task the test completes
    /// on demand (so seed/reconcile timing relative to buffered deltas is deterministic).
    /// </summary>
    private sealed class SeedStub
    {
        private readonly ConcurrentQueue<TaskCompletionSource<IReadOnlyList<AlarmStateChanged>>> _pending = new();
        private int _callCount;

        public int CallCount => Volatile.Read(ref _callCount);

        public Task<IReadOnlyList<AlarmStateChanged>> Seed(CancellationToken ct)
        {
            Interlocked.Increment(ref _callCount);
            var tcs = new TaskCompletionSource<IReadOnlyList<AlarmStateChanged>>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            ct.Register(() => tcs.TrySetCanceled());
            _pending.Enqueue(tcs);
            return tcs.Task;
        }

        /// <summary>Completes the oldest not-yet-completed seed call with the given rows.</summary>
        public void CompleteNext(params AlarmStateChanged[] rows)
        {
            SpinWait.SpinUntil(() => _pending.TryPeek(out _), TimeSpan.FromSeconds(3));
            if (_pending.TryDequeue(out var tcs))
                tcs.TrySetResult(rows);
            else
                throw new InvalidOperationException("No pending seed to complete.");
        }

        /// <summary>Faults the oldest not-yet-completed seed call (simulates a whole-fan-out failure).</summary>
        public void FaultNext()
        {
            SpinWait.SpinUntil(() => _pending.TryPeek(out _), TimeSpan.FromSeconds(3));
            if (_pending.TryDequeue(out var tcs))
                tcs.TrySetException(new InvalidOperationException("seed boom"));
            else
                throw new InvalidOperationException("No pending seed to fault.");
        }
    }

    private sealed class PublishSink
    {
        private readonly object _lock = new();
        public List<IReadOnlyList<AlarmStateChanged>> Snapshots { get; } = new();

        public void Publish(IReadOnlyList<AlarmStateChanged> snapshot, bool streamLive)
        {
            lock (_lock) { Snapshots.Add(snapshot); StreamLive = streamLive; }
        }

        /// <summary>Liveness reported with the most recent publish.</summary>
        public bool StreamLive { get; private set; }

        public IReadOnlyList<AlarmStateChanged>? Latest
        {
            get { lock (_lock) { return Snapshots.Count == 0 ? null : Snapshots[^1]; } }
        }

        public int Count { get { lock (_lock) { return Snapshots.Count; } } }
    }

    private sealed record SiteSub(
        string CorrelationId, Action<AlarmStateChanged> OnAlarm, Action<Exception> OnError,
        Action OnCompleted, CancellationToken Ct, Action? OnConnected);

    private sealed class MockSiteAlarmStreamClient : SiteStreamGrpcClient
    {
        private readonly object _lock = new();
        private readonly List<SiteSub> _subs = new();
        private readonly List<string> _unsubscribed = new();

        public List<SiteSub> Subs { get { lock (_lock) { return _subs.ToList(); } } }
        public List<string> Unsubscribed { get { lock (_lock) { return _unsubscribed.ToList(); } } }

        public MockSiteAlarmStreamClient() : base() { }

        /// <summary>When false the stream never reports connected — the site accepted the TCP
        /// call but never answered, so the aggregator's connect-driven re-seed must not run.</summary>
        public bool AutoConnect { get; set; } = true;

        public override Task SubscribeSiteAsync(
            string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> onError,
            Action onCompleted, CancellationToken ct, Action? onConnected = null)
        {
            lock (_lock)
            {
                _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, onCompleted, ct, onConnected));
            }
            // A healthy site accepts the subscription immediately (it flushes response headers
            // as soon as its relay is attached), which is the aggregator's connect signal.
            if (AutoConnect) onConnected?.Invoke();
            var tcs = new TaskCompletionSource();
            ct.Register(() => tcs.TrySetResult());
            return tcs.Task; // never completes until cancelled (simulates a live stream)
        }

        public override void Unsubscribe(string correlationId)
        {
            lock (_lock) { _unsubscribed.Add(correlationId); }
        }
    }

    private sealed class MockSiteAlarmStreamClientFactory : SiteStreamGrpcClientFactory
    {
        private readonly ConcurrentDictionary<string, MockSiteAlarmStreamClient> _byEndpoint = new();

        public MockSiteAlarmStreamClientFactory()
            : base(Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance) { }

        /// <summary>Applied to every client this factory hands out (set before the actor starts).</summary>
        public bool AutoConnect { get; set; } = true;

        public MockSiteAlarmStreamClient ClientFor(string endpoint) =>
            _byEndpoint.GetOrAdd(endpoint, _ => new MockSiteAlarmStreamClient { AutoConnect = AutoConnect });

        public override SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint)
            => ClientFor(grpcEndpoint);

        public override SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint)
            => _byEndpoint.TryGetValue(grpcEndpoint, out var c) ? c : null;
    }

    private (IActorRef Actor, SeedStub Seed, PublishSink Sink, MockSiteAlarmStreamClientFactory Factory) CreateActor(
        TimeSpan? reconcileInterval = null, TimeSpan? publishCoalesce = null, bool autoConnect = true)
    {
        var seed = new SeedStub();
        var sink = new PublishSink();
        var factory = new MockSiteAlarmStreamClientFactory { AutoConnect = autoConnect };

        var props = Props.Create(() => new SiteAlarmAggregatorActor(
            SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB,
            reconcileInterval ?? TimeSpan.FromMinutes(10),
            publishCoalesce ?? TimeSpan.Zero,
            TestReconnectDelay,
            TestStabilityWindow,
            0.0)); // no reconcile jitter in tests — determinism

        var actor = Sys.ActorOf(props);
        return (actor, seed, sink, factory);
    }

    private static AlarmStateChanged Alarm(
        string alarmName, string sourceRef, int priority, DateTimeOffset ts,
        bool placeholder = false, string instance = Instance) =>
        new(instance, alarmName, AlarmState.Active, priority, ts)
        {
            SourceReference = sourceRef,
            IsConfiguredPlaceholder = placeholder,
            Kind = placeholder ? AlarmKind.NativeOpcUa : AlarmKind.Computed
        };

    private static AlarmStateChanged? Find(IReadOnlyList<AlarmStateChanged>? rows, string alarmName, string sourceRef) =>
        rows?.FirstOrDefault(r => r.AlarmName == alarmName && r.SourceReference == sourceRef);

    // ── Tests ───────────────────────────────────────────────────────────────────

    [Fact]
    public void Delta_Arriving_Before_Seed_Completes_Is_Buffered_And_Applied_Exactly_Once()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        // Live delta arrives DURING the seed window (new key, not in the seed).
        var t1 = DateTimeOffset.UtcNow;
        actor.Tell(Alarm("Overheat", "", 900, t1));

        // Still buffering — nothing published yet.
        Assert.Equal(0, sink.Count);

        // Seed completes with a DIFFERENT alarm.
        seed.CompleteNext(Alarm("PumpFault", "", 500, t1.AddSeconds(-1)));

        AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3));
        var latest = sink.Latest!;
        Assert.NotNull(Find(latest, "PumpFault", "")); // seeded
        var overheat = Find(latest, "Overheat", "");   // buffered gap-window delta survived
        Assert.NotNull(overheat);
        Assert.Single(latest, r => r.AlarmName == "Overheat"); // exactly once
    }

    [Fact]
    public void Buffered_Delta_Older_Than_Seed_Entry_Is_Dropped()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var tSeed = DateTimeOffset.UtcNow;
        // Buffered delta for the SAME key but OLDER than the seed entry.
        actor.Tell(Alarm("PumpFault", "", 100, tSeed.AddSeconds(-1)));
        seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed));

        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));
        // Give a beat to ensure no late (dropped) delta sneaks in.
        Thread.Sleep(150);
        var latest = sink.Latest!;
        var pumpFault = Find(latest, "PumpFault", "");
        Assert.NotNull(pumpFault);
        Assert.Equal(500, pumpFault!.Priority); // seed value kept; older buffered delta dropped
    }

    [Fact]
    public void Buffered_Delta_Newer_Than_Seed_Entry_Is_Applied()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var tSeed = DateTimeOffset.UtcNow;
        actor.Tell(Alarm("PumpFault", "", 900, tSeed.AddSeconds(1))); // newer than seed
        seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed));

        AwaitCondition(() => sink.Latest is { } l && Find(l, "PumpFault", "")?.Priority == 900,
            TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void Dedup_By_AlarmKey_Distinguishes_SourceReference()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext(); // empty seed
        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        // Same instance + alarm name, different source reference → two distinct rows.
        actor.Tell(Alarm("LevelAlarm", "Tank01.Level.HiHi", 700, t));
        actor.Tell(Alarm("LevelAlarm", "Tank01.Level.LoLo", 300, t));

        AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3));
        var latest = sink.Latest!;
        Assert.NotNull(Find(latest, "LevelAlarm", "Tank01.Level.HiHi"));
        Assert.NotNull(Find(latest, "LevelAlarm", "Tank01.Level.LoLo"));
    }

    [Fact]
    public void Same_Key_Live_Update_Replaces_In_Place()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();
        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        actor.Tell(Alarm("PumpFault", "", 500, t));
        AwaitCondition(() => Find(sink.Latest, "PumpFault", "")?.Priority == 500, TimeSpan.FromSeconds(3));

        // Newer update for the same key replaces (not duplicates).
        actor.Tell(Alarm("PumpFault", "", 800, t.AddSeconds(1)));
        AwaitCondition(() => Find(sink.Latest, "PumpFault", "")?.Priority == 800, TimeSpan.FromSeconds(3));
        Assert.Single(sink.Latest!); // still exactly one row for the key
    }

    [Fact]
    public void Seeded_Native_Alarm_And_Its_Live_Delta_Collapse_Onto_One_Key()
    {
        // Identity parity across the two paths (snapshot seed vs live stream): a native
        // alarm seeded with a non-empty SourceReference and a later live delta for the
        // SAME (instance, alarmName, sourceRef) must collapse to ONE in-place-updated row,
        // not a ghost duplicate. Both paths key on SourceReference, so they must agree.
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        seed.CompleteNext(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 300, t));
        AwaitCondition(() => Find(sink.Latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi") is not null,
            TimeSpan.FromSeconds(3));

        // Live delta for the exact same identity, newer → updates in place.
        actor.Tell(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 800, t.AddSeconds(1)));

        AwaitCondition(
            () => Find(sink.Latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi")?.Priority == 800,
            TimeSpan.FromSeconds(3));
        Assert.Single(sink.Latest!); // exactly one row — no seed/live ghost duplicate
    }

    [Fact]
    public void Placeholder_Seeded_Coexists_With_Live_Real_Alarm_And_Is_Not_Wiped()
    {
        var (actor, seed, sink, _) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        // Seed carries a configured-placeholder row (binding with no active conditions).
        var placeholder = Alarm("Motor1.MotorAlarms", "", 0, t, placeholder: true);
        seed.CompleteNext(placeholder);
        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));

        // A live REAL condition for the same instance arrives (distinct AlarmKey).
        actor.Tell(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 800, t.AddSeconds(1)));

        AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3));
        var latest = sink.Latest!;
        // Placeholder still present (not wiped by the real alarm under a different key).
        var ph = Find(latest, "Motor1.MotorAlarms", "");
        Assert.NotNull(ph);
        Assert.True(ph!.IsConfiguredPlaceholder);
        Assert.NotNull(Find(latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi"));
    }

    [Fact]
    public void GrpcError_Flips_Node_Reconnects_And_ReSeeds()
    {
        var (_, seed, sink, factory) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext(); // initial seed done
        // Wait for the seed to actually be APPLIED (published) so the re-seed on error is
        // not skipped as "fan-out already in flight".
        AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3));

        // Fail the NodeA stream.
        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("NodeA down"));

        // Reconnect must reach NodeB, AND a re-seed fan-out must have been triggered.
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5));
        // The failed NodeA stream was unsubscribed (relay released, not left zombie).
        Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).Unsubscribed);
    }

    [Fact]
    public void Periodic_Reconcile_Authoritatively_Replaces_Stale_Rows()
    {
        var (_, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(250));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        seed.CompleteNext(Alarm("PumpFault", "", 500, t)); // seed has a row
        AwaitCondition(() => Find(sink.Latest, "PumpFault", "") is not null, TimeSpan.FromSeconds(3));

        // Reconcile tick fires → seed #2 (e.g. instance was disabled) returns EMPTY.
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
        seed.CompleteNext(); // empty

        // The stale PumpFault row must disappear (authoritative replace, not merge).
        AwaitCondition(() => sink.Latest is { } l && l.Count == 0, TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void Failed_Initial_Seed_Does_Not_Publish_But_A_Later_Reconcile_Recovers()
    {
        var (_, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(250));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        // The initial seed fan-out throws as a whole.
        seed.FaultNext();

        // A failed INITIAL seed must not publish (page keeps its poll fallback — not live yet).
        Thread.Sleep(200);
        Assert.Equal(0, sink.Count);

        // The periodic reconcile is the backstop: it re-runs the fan-out (flag was cleared
        // on failure), completes, and the actor publishes → now live.
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
        seed.CompleteNext(Alarm("PumpFault", "", 500, DateTimeOffset.UtcNow));
        AwaitCondition(() => Find(sink.Latest, "PumpFault", "") is not null, TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void Stream_GivenUp_After_MaxRetries_Reopens_On_Reconcile_Tick()
    {
        // After the retry budget is exhausted the live stream is left down, but the
        // aggregator is NOT stopped — the next reconcile tick self-heals it (resets the
        // budget + reopens) so a sustained site outage never permanently kills the feed.
        var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(300));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));

        // Drive initial + 3 retries (each flips node) to exhaust the budget.
        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("1"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeB).Subs[0].OnError(new Exception("2"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5));

        int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count;
        var before = TotalSubs(); // 4

        // Fourth error exceeds MaxRetries → stream given up (no immediate reopen).
        factory.ClientFor(GrpcNodeB).Subs[1].OnError(new Exception("4"));

        // The reconcile tick reopens the stream (self-heal).
        AwaitCondition(() => TotalSubs() > before, TimeSpan.FromSeconds(3));
    }

    // ── R2 T11: queued re-seed after reconnect + stream-generation stamp (N7.1/N7.2) ──

    [Fact]
    public void ReseedRequestedDuringInFlightFanout_IsQueued_NotDropped()
    {
        var (actor, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.Zero);
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();                       // initial seed done (CallCount 1)
        AwaitAssert(() => Assert.Equal(1, sink.Count));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));

        // The first tick after a fan-out is deliberately skipped (WP2.3: the initial seed
        // already covered this window); the second one actually fans out.
        actor.Tell(new RunReconcile());            // skipped — consumes the "already fanned out" flag
        actor.Tell(new RunReconcile());            // reconcile fan-out now in flight (CallCount 2)
        AwaitAssert(() => Assert.Equal(2, seed.CallCount));

        // Stream error while the reconcile is in flight: the reconnect that follows
        // reconnects (the mock accepts immediately) and its connect-driven re-seed must not
        // be silently swallowed. Pre-fix: StartFanout no-ops and CallCount stays 2 until the
        // next 60s reconcile tick.
        factory.ClientFor(GrpcNodeA).Subs.Last().OnError(new Exception("stream fault"));

        seed.CompleteNext();                       // finish the in-flight reconcile
        AwaitAssert(() => Assert.Equal(3, seed.CallCount)); // queued re-seed ran immediately after
    }

    [Fact]
    public void LateErrorFromAPreviousStreamGeneration_IsIgnored()
    {
        var (_, seed, _, factory) = CreateActor(publishCoalesce: TimeSpan.Zero);
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));

        int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count;

        var firstSub = factory.ClientFor(GrpcNodeA).Subs.Single();
        firstSub.OnError(new Exception("real fault"));      // gen 1 error → flip, reconnect
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); // gen 2 stream open
        AwaitAssert(() => Assert.Equal(2, TotalSubs()));

        firstSub.OnError(new Exception("late zombie fault")); // stale gen-1 error races in
        // Pre-fix this burns retry budget and opens a THIRD stream; post-fix it is ignored.
        Thread.Sleep(400);
        Assert.Equal(2, TotalSubs());
    }

    // ── WP1.1: graceful (status OK) stream completion is a reconnect trigger ──

    [Fact]
    public void GracefulStreamCompletion_ReopensOnReconcileTick_OnTheSameNode()
    {
        // The site ends every stream with OK at its 4h max lifetime. Pre-fix the client's
        // read loop just finished, nothing was told to the actor, and the site's alarm feed
        // stayed silently dead until the central node restarted.
        var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(300));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();

        factory.ClientFor(GrpcNodeA).Subs[0].OnCompleted();

        // Reopened by the reconcile tick, on the SAME node — a clean close is not a fault,
        // so there is nothing to fail over from.
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(3));
        Assert.Empty(factory.ClientFor(GrpcNodeB).Subs);
        // The finished stream was released, so the site keeps no zombie relay actor.
        Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).Unsubscribed);
    }

    [Fact]
    public void GracefulStreamCompletion_NeitherSpendsNorRefunds_TheErrorRetryBudget()
    {
        // Reconcile is far away; reopens are driven explicitly so the budget can be observed.
        var (actor, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();

        // Spend the whole error budget (MaxRetries = 3), each error flipping the node.
        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("1"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeB).Subs[0].OnError(new Exception("2"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5));

        // A graceful completion reopens without flipping the node (budget not spent) …
        factory.ClientFor(GrpcNodeB).Subs[1].OnCompleted();
        actor.Tell(new RunReconcile());
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 3, TimeSpan.FromSeconds(5));
        Assert.Equal(2, factory.ClientFor(GrpcNodeA).Subs.Count);

        int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count;
        var before = TotalSubs();

        // … and without refunding it either: the next error is the 4th, so it exceeds
        // MaxRetries and the stream is given up rather than reconnected.
        factory.ClientFor(GrpcNodeB).Subs[2].OnError(new Exception("4"));
        Thread.Sleep(400);
        Assert.Equal(before, TotalSubs());
    }

    [Fact]
    public void LateCompletionFromAPreviousStreamGeneration_IsIgnored()
    {
        var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();

        var firstSub = factory.ClientFor(GrpcNodeA).Subs.Single();
        firstSub.OnError(new Exception("real fault"));  // gen 1 dies → gen 2 opens on NodeB
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));

        // A completion racing out of the dead gen-1 stream must not tear down the live
        // gen-2 stream (which would go deltaless until the next reconcile tick).
        firstSub.OnCompleted();
        Thread.Sleep(300);
        Assert.DoesNotContain("corr-1", factory.ClientFor(GrpcNodeB).Unsubscribed);
        Assert.Single(factory.ClientFor(GrpcNodeB).Subs);
    }

    // ── R2 T10: live-delta publish coalescing (N6) ──

    [Fact]
    public void DeltaBurst_IsCoalesced_IntoFewPublishes_ContainingEveryRow()
    {
        var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.FromMilliseconds(150));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext(); // empty initial seed
        AwaitAssert(() => Assert.Equal(1, sink.Count)); // seed publish, immediate

        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        var sub = factory.ClientFor(GrpcNodeA).Subs.Single();
        var now = DateTimeOffset.UtcNow;
        for (var i = 0; i < 50; i++)
            sub.OnAlarm(Alarm($"A{i}", "", 900, now)); // 50 distinct keys, one burst

        AwaitAssert(() =>
        {
            Assert.Equal(50, sink.Latest!.Count);   // nothing lost
            Assert.InRange(sink.Count, 2, 4);       // pre-fix: 51 publishes (1 seed + 1 per delta)
        }, TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void ZeroCoalesce_PreservesLegacyPerDeltaPublish()
    {
        var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.Zero);
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();
        AwaitAssert(() => Assert.Equal(1, sink.Count));

        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        var sub = factory.ClientFor(GrpcNodeA).Subs.Single();
        var now = DateTimeOffset.UtcNow;
        sub.OnAlarm(Alarm("A1", "", 900, now));
        sub.OnAlarm(Alarm("A2", "", 900, now));
        AwaitAssert(() => Assert.Equal(3, sink.Count)); // 1 seed + 1 per delta
    }

    [Fact]
    public void Stop_Message_TearsDown_Grpc_And_Stops_Actor()
    {
        var (actor, seed, _, factory) = CreateActor();
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();

        Watch(actor);
        actor.Tell(new StopSiteAlarmAggregator());

        ExpectTerminated(actor, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Unsubscribed.Contains("corr-1"),
            TimeSpan.FromSeconds(3));
    }

    // -- WP2.3: seed once per SUCCESSFUL (re)connect, and stream-truthful liveness --

    [Fact]
    public void ReconnectAttempts_DoNotEachReSeed_OnlyTheSuccessfulConnectDoes()
    {
        // Finding #10's per-attempt re-fan-out: every reconnect ATTEMPT used to kick a
        // whole-site snapshot fan-out - against a site that is, by definition of the
        // reconnect, unreachable. Now the re-seed is owed to the connect that succeeds.
        var (_, seed, _, factory) = CreateActor(
            reconcileInterval: TimeSpan.FromMinutes(10), autoConnect: false);
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); // initial seed
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();

        // Three failed attempts, each flipping the node - none of them may re-seed.
        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("1"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeB).Subs[0].OnError(new Exception("2"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5));

        Thread.Sleep(300);
        Assert.Equal(1, seed.CallCount); // pre-fix: 4

        // The stream finally comes up - exactly ONE re-seed, on the connect.
        factory.ClientFor(GrpcNodeB).Subs[1].OnConnected!();
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5));
        seed.CompleteNext();
        Thread.Sleep(300);
        Assert.Equal(2, seed.CallCount); // and no second one
    }

    [Fact]
    public void ReconcileTick_IsSkipped_WhenAFanoutAlreadyRanInTheWindow()
    {
        // The reconcile is a BACKSTOP, not an unconditional 60s whole-site snapshot: a
        // window already covered by a seed costs nothing. Staleness stays bounded because
        // the skip consumes the flag, so the next tick always fans out.
        var (actor, seed, _, _) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();
        Thread.Sleep(200);

        actor.Tell(new RunReconcile());   // skipped - the initial seed covered this window
        Thread.Sleep(300);
        Assert.Equal(1, seed.CallCount);

        actor.Tell(new RunReconcile());   // and the next one runs
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void UnchangedReconcileSnapshot_DoesNotRepublish()
    {
        // A reconcile that finds nothing changed used to wake every viewer's render path
        // once a minute regardless. It now publishes as a diff.
        var (actor, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        var t = DateTimeOffset.UtcNow;
        seed.CompleteNext(Alarm("PumpFault", "", 500, t));
        AwaitAssert(() => Assert.Equal(1, sink.Count));

        actor.Tell(new RunReconcile()); // skipped (seed covered the window)
        actor.Tell(new RunReconcile()); // real fan-out
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
        seed.CompleteNext(Alarm("PumpFault", "", 500, t)); // identical snapshot

        Thread.Sleep(300);
        Assert.Equal(1, sink.Count); // no second publish

        // A genuinely changed snapshot still publishes.
        actor.Tell(new RunReconcile());
        actor.Tell(new RunReconcile());
        AwaitCondition(() => seed.CallCount == 3, TimeSpan.FromSeconds(3));
        seed.CompleteNext(Alarm("PumpFault", "", 900, t.AddSeconds(1)));
        AwaitAssert(() => Assert.Equal(2, sink.Count));
    }

    [Fact]
    public void StreamCompletion_DropsLiveness_AndTheReopenRestoresIt()
    {
        // WP2.3 carried residual: a completed (or given-up) stream kept reporting live
        // until the next reconcile publish, so the page grafted a freezing snapshot over
        // fresh poll data.
        // Reconcile is held far away so the drop can be observed before the reopen
        // restores liveness; the reopen is then driven explicitly.
        var (actor, seed, sink, factory) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();
        AwaitCondition(() => sink.Count >= 1 && sink.StreamLive, TimeSpan.FromSeconds(3));

        // The site ends the stream cleanly (its 4h max lifetime).
        factory.ClientFor(GrpcNodeA).Subs[0].OnCompleted();

        // Liveness drops IMMEDIATELY - not at the next reconcile.
        AwaitCondition(() => !sink.StreamLive, TimeSpan.FromSeconds(3));

        // The reconcile tick reopens; the connect restores liveness.
        actor.Tell(new RunReconcile());
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5));
        AwaitCondition(() => sink.StreamLive, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void StreamFault_DropsLiveness_Immediately()
    {
        var (_, seed, sink, factory) = CreateActor(
            reconcileInterval: TimeSpan.FromMinutes(10), autoConnect: false);
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        factory.ClientFor(GrpcNodeA).Subs[0].OnConnected!();
        seed.CompleteNext();
        AwaitCondition(() => sink.StreamLive, TimeSpan.FromSeconds(3));

        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("site gone"));
        AwaitCondition(() => !sink.StreamLive, TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void FailedSeed_IsRetried_WithBackoff_BeforeTheNextReconcile()
    {
        // The seed leg had no backoff: a failing fan-out simply waited a full reconcile
        // interval. Now it retries on its own timer (reconnectDelay, doubling).
        var (_, seed, _, _) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));

        seed.FaultNext();
        // TestReconnectDelay is 50 ms, so the first retry lands far inside the 10-minute
        // reconcile interval - pre-fix nothing would run until that interval elapsed.
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5));
        seed.FaultNext();
        AwaitCondition(() => seed.CallCount == 3, TimeSpan.FromSeconds(5));
    }
}