ScadaBridge/archreview/plans/PLAN-R2-02-communication-store-and-forward.md

PLAN-R2-02 — Communication & Store-and-Forward Round-2 Fix Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers-extended-cc:executing-plans to implement this plan task-by-task.

Goal: Fix every NEW finding in archreview/02-communication-store-and-forward.md (round 2, dated 2026-07-12, on main @ 8c888f13) — the Critical resync predicate inversion that lets a routine rolling restart wipe the delivering node's live S&F buffer (N1, which also closes the round-1 partial: the leader-vs-oldest helper was never swapped into SiteCommunicationActor/SiteReplicationActor), the High frame-size-undeliverable resync snapshot (N2), the _sweepTask clobber that defeats the shutdown drain (N3), live-alarm delta coalescing (N6), and the Low hygiene items (N4, N5, N7, N8, N9).

Architecture: The fixes stay inside the existing seams. A single oldest-Up active-node predicate (ActiveNodeEvaluator, new in Communication, delegated to by the Host's ClusterActivityEvaluator) replaces both surviving leader+Up defaults, and the Host wires the same IClusterNodeProvider.SelfIsPrimary delegate that already gates the S&F delivery sweep into SiteReplicationActor and SiteCommunicationActor — one definition of "active" governs delivery, resync-request, resync-answer, and heartbeat IsActive. The anti-entropy snapshot becomes a chunked, byte-budgeted, ack-confirmed protocol (additive messages; the legacy monolithic SfBufferSnapshot handler is retained for rolling compat). The sweep handoff publishes _sweepTask only when the CAS is won. The live alarm aggregator gains a publish-coalescing timer, a queued re-seed, a stream-generation stamp, and instance-injected test seams; SiteStreamGrpcClientFactory.RemoveSiteAsync gets its production caller on site delete; single-endpoint sites can go live.

Tech Stack: C#/.NET 10, Akka.NET (TestKit.Xunit2, Cluster, ClusterClient), Microsoft.Data.Sqlite, Grpc.Net, xUnit + NSubstitute. Build: dotnet build ZB.MOM.WW.ScadaBridge.slnx. Test per-project: dotnet test tests/<project> — targeted --filter runs only, never the full suite mid-plan.

Cross-plan MUTEX (binding)

PLAN-R2-07 also touches SiteAlarmLiveCacheService.cs (sticky IsLive) and AlarmSummary.razor. Serialization rule: Task 10 (delta coalescing), Task 12 (aggregator ctor seams), and Task 15 (single-endpoint resolve) must NOT run concurrently with PLAN-R2-07's sticky-IsLive / AlarmSummary tasks. Tasks 10/12/15 edit SiteAlarmAggregatorActor.cs + the Props.Create call and ResolveSiteAsync inside SiteAlarmLiveCacheService.cs — the exact file R2-07's sticky-IsLive task rewrites, and Task 10 changes the publish cadence that sticky-IsLive semantics observe. Whichever plan starts its live-alarm slice first finishes it (through commit) before the other begins; coordinate via the master tracker. All other tasks in this plan are file-disjoint from R2-07. AlarmSummary.razor is never edited here (page-side fixes belong to PLAN-R2-07).

Findings Coverage

Report finding	Severity	Task(s)
N1: Resync authority uses leader predicate while delivery uses oldest-Up — rolling restart of the lower-address node wipes the delivering node's live buffer	Critical	1, 2, 3, 4 (2 is the mandated failing-first divergence repro; 3 includes the belt-and-braces apply-time re-check)
Round-1 partial: oldest-Up helper shipped (PLAN-01) but never swapped into SiteCommunicationActor.DefaultIsActiveCheck / SiteReplicationActor.DefaultIsActive	Low → materially worse	Subsumed by N1's fix — 1 (shared evaluator), 3 (SiteReplicationActor), 4 (SiteCommunicationActor)
N2: Resync snapshot rides ONE Akka remoting message — exceeds the 128 000-byte default frame for any realistic backlog, silently undeliverable, no retry, no telemetry	High	5, 6, 7 (chunked byte-budgeted protocol + standby assembly + ack-based delivery confirmation + telemetry)
N3: _sweepTask clobbered by every timer tick / TriggerSweep — StopAsync awaits a no-op while the real sweep is mid-delivery	Medium	8
N4: SweepBatchLimit / SweepTargetParallelism missing from the eager options validator	Low	9
N5: Resync snapshot vs in-flight replication op race leaves orphan rows on the standby	Low	6 (explicit doc comment on the chunk-apply path so nobody "fixes" it into something worse — code change deliberately none, per report)
N6: No live-delta coalescing — every alarm delta copies the whole site cache and fans out to every viewer circuit	Medium	10 (MUTEX with PLAN-R2-07 — see rule above)
N7.1: Reconnect re-seed skipped when a reconcile fan-out is already in flight (stale up to 60 s)	Low	11
N7.2: No stream-generation stamp on GrpcAlarmStreamError — a late error from the previous stream burns retry budget	Low	11
N7.3: ReconnectDelay/StabilityWindow are process-global mutable statics	Low	12
N8: RemoveSiteAsync has no production caller — deleted site's channels leak until restart	Low	13
N8 (second half): "lives only on the active central node" is aspirational — standby browsing starts a second aggregator	Low	14 (documented acceptance — read-only, bounded, linger-reaped; doc claim corrected rather than gating SetActorSystem, matching the report's "harmless but contradicting the doc" framing)
N9: A site with only one configured gRPC endpoint can never go live	Low	15
Alarm Summary page-side fixes (poll/live interplay, render path)	—	Not this plan — owned by PLAN-R2-07 (referenced here only for the MUTEX rule)

---

Task 1: Shared oldest-Up active-node evaluator, reachable from Communication and SiteRuntime

Classification: standard Estimated implement time: ~5 min Parallelizable with: 2, 8, 9, 10, 13, 14 Files:

• Create: src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs
• Modify: src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs (lines 20-32 — SelfIsOldest delegates to the new evaluator)
• Test: tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs (create)

Design decision — new home in Communication, Host delegates: PLAN-01's ClusterActivityEvaluator (Host/Health/ClusterActivityEvaluator.cs:14) declares itself "THE single definition of active node", but it lives in Host, which Communication and SiteRuntime cannot reference (Host references them). That is why the two leader+Up defaults were never swapped (round-1 partial). The logic moves to Communication (referenced by both SiteRuntime — see ZB.MOM.WW.ScadaBridge.SiteRuntime.csproj:39 — and Host); Host's evaluator keeps its public API and becomes a one-line delegate, so AkkaClusterNodeProvider.SelfIsPrimary (Host/Health/AkkaClusterNodeProvider.cs:29-38) and every existing Host test keep working against the same single implementation.

1. Write the failing test (cluster-provider harness copied from tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ClusterActivityEvaluatorTests.cs — single-node self-seeding cluster on a free port):

// tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs
using System.Net;
using System.Net.Sockets;
using Akka.Actor;
using Akka.Configuration;
using Xunit;
using ZB.MOM.WW.ScadaBridge.Communication.ClusterState;

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

public class ActiveNodeEvaluatorTests : IAsyncLifetime
{
    private ActorSystem? _system;

    public async Task InitializeAsync()
    {
        var port = FreePort();
        var config = ConfigurationFactory.ParseString($@"
akka {{
    actor.provider = cluster
    remote.dot-netty.tcp {{ hostname = ""127.0.0.1"", port = {port} }}
    cluster {{
        seed-nodes = [""akka.tcp://ane-test@127.0.0.1:{port}""]
        roles = [""site-x""]
        min-nr-of-members = 1
    }}
}}");
        _system = ActorSystem.Create("ane-test", config);
        var cluster = Akka.Cluster.Cluster.Get(_system);
        var deadline = DateTime.UtcNow.AddSeconds(20);
        while (cluster.SelfMember.Status != Akka.Cluster.MemberStatus.Up && DateTime.UtcNow < deadline)
            await Task.Delay(100);
    }

    public async Task DisposeAsync() { if (_system != null) await _system.Terminate(); }

    [Fact]
    public void SelfIsOldestUp_SoleUpMember_ReturnsTrue()
    {
        var cluster = Akka.Cluster.Cluster.Get(_system!);
        Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(cluster));
        Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(cluster, "site-x"));
    }

    [Fact]
    public void SelfIsOldestUp_RoleNotHeld_ReturnsFalse()
    {
        var cluster = Akka.Cluster.Cluster.Get(_system!);
        Assert.False(ActiveNodeEvaluator.SelfIsOldestUp(cluster, "role-nonexistent"));
    }

    private static int FreePort()
    {
        using var l = new TcpListener(IPAddress.Loopback, 0);
        l.Start();
        return ((IPEndPoint)l.LocalEndpoint).Port;
    }
}

2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter ActiveNodeEvaluatorTests — expect FAIL (compile error: ActiveNodeEvaluator does not exist).
3. Implement src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs (verbatim move of the oldest-Up logic from ClusterActivityEvaluator.SelfIsOldest):

using Akka.Cluster;

namespace ZB.MOM.WW.ScadaBridge.Communication.ClusterState;

/// <summary>
/// THE single definition of "active node" (review 01 [High]; review 02 round 2 [Critical] N1):
/// a node is active when it is the OLDEST Up member (optionally within a role scope) — i.e.
/// the member the ClusterSingletonManager places singletons on. Cluster LEADERSHIP (lowest
/// address) is an Akka-internal concept that diverges from singleton placement permanently
/// once the original first node restarts and rejoins; every product-level active/standby
/// decision must use this evaluator, never <c>cluster.State.Leader</c>.
/// <para>
/// Lives in Communication (not Host) so BOTH <c>SiteCommunicationActor</c> and
/// <c>SiteReplicationActor</c> can default to it — Host cannot be referenced from either.
/// The Host's <c>ClusterActivityEvaluator.SelfIsOldest</c> delegates here, so the S&amp;F
/// delivery gate (<c>IClusterNodeProvider.SelfIsPrimary</c>), the resync authority checks,
/// and the heartbeat IsActive stamp all share one implementation.
/// </para>
/// </summary>
public static class ActiveNodeEvaluator
{
    /// <summary>True when self is Up and no other Up member (in the role scope) is older.</summary>
    /// <param name="cluster">The Akka cluster to evaluate.</param>
    /// <param name="role">Optional role scope; when set, only members with this role are considered.</param>
    /// <returns><c>true</c> when self is Up and the oldest Up member in the role scope.</returns>
    public static bool SelfIsOldestUp(Cluster cluster, string? role = null)
    {
        var self = cluster.SelfMember;
        if (self.Status != MemberStatus.Up)
            return false;
        if (role != null && !self.HasRole(role))
            return false;

        return cluster.State.Members
            .Where(m => m.Status == MemberStatus.Up)
            .Where(m => role == null || m.HasRole(role))
            .All(m => m.UniqueAddress.Equals(self.UniqueAddress) || self.IsOlderThan(m));
    }
}

And in Host/Health/ClusterActivityEvaluator.cs, replace the SelfIsOldest body (keep OldestUpMember unchanged):

    /// <summary>True when self is Up and no other Up member (in the role scope) is older.
    /// Delegates to the shared <see cref="Communication.ClusterState.ActiveNodeEvaluator"/> —
    /// one implementation for the delivery gate, the resync authority checks, and the
    /// heartbeat IsActive stamp (review 02 round 2, N1).</summary>
    public static bool SelfIsOldest(Cluster cluster, string? role = null) =>
        Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(cluster, role);

(Full namespace: ZB.MOM.WW.ScadaBridge.Communication.ClusterState — add a using alias if the shorthand does not resolve.) 4. Run the filter — expect PASS. Regression: dotnet test tests/ZB.MOM.WW.ScadaBridge.Host.Tests --filter ClusterActivityEvaluatorTests — expect PASS (delegation preserves behavior). 5. Commit: git add src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs && git commit -m "fix(comm): shared oldest-Up ActiveNodeEvaluator in Communication; Host evaluator delegates (plan R2-02 T1)"

Task 2: Two-node divergence repro — leader≠oldest wipes the delivering node's buffer (failing first)

Classification: high-risk (real two-node cluster; the mandated N1 regression repro) Estimated implement time: ~5 min Parallelizable with: 1, 8, 9, 10, 13, 14, 15 Files:

• Modify: tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs (StartAsync gains optional explicit ports — additive)
• Test: tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs (create)

This test is written BEFORE the fix and must FAIL on unfixed code — it reproduces the report's N1 sequence. The report narrates it as a rolling restart of the lower-address node; the essence is older-but-higher-address vs younger-but-lower-address, which the test produces directly (deterministically, no crash/restart needed) by starting the higher-port node first: the first-started (higher-address) node is oldest (owns delivery), the second-started (lower-address) node is leader. Pre-fix, SiteReplicationActor.DefaultIsActive (SiteReplicationActor.cs:190-198, leader+Up) then makes the delivering node request a resync and apply the joiner's stale/empty snapshot via ReplaceAllAsync — the buffer wipe. Post-fix (Task 3), the oldest node answers and the joiner applies: the buffered row survives on the oldest node AND appears on the joiner.

1. Add the additive fixture parameters (TwoNodeClusterFixture.StartAsync, currently StartAsync(string role = "Central", TimeSpan? stableAfter = null)):

    public static async Task<TwoNodeClusterFixture> StartAsync(
        string role = "Central", TimeSpan? stableAfter = null,
        int? portA = null, int? portB = null)
    {
        var f = new TwoNodeClusterFixture();
        f.PortA = portA ?? GetFreeTcpPort();
        f.PortB = portB ?? GetFreeTcpPort();
        // ... rest unchanged

2. Write the failing test:

// tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs
using Akka.Actor;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
using ZB.MOM.WW.ScadaBridge.StoreAndForward;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;

namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster;

/// <summary>
/// N1 regression (review 02 round 2, Critical): the resync authority must use the same
/// oldest-Up predicate as the S&F delivery gate. Divergence scenario = the delivering node
/// is OLDEST but not LEADER (leader = lowest address), the exact state a rolling restart of
/// the lower-address node produces. Pre-fix the delivering node requests a resync from the
/// stale peer and ReplaceAllAsync wipes its live buffer.
/// </summary>
public class SfBufferResyncPredicateTests
{
    [Fact]
    public async Task OldestButNotLeaderNode_KeepsItsBuffer_AndSeedsTheJoiner()
    {
        // Two explicit ports, deliberately assigned so the FIRST-started (oldest,
        // delivering) node has the HIGHER address → the second node is cluster leader.
        var p1 = TwoNodeClusterFixture.GetFreeTcpPort();
        var p2 = TwoNodeClusterFixture.GetFreeTcpPort();
        var (portHigh, portLow) = p1 > p2 ? (p1, p2) : (p2, p1);

        await using var fixture = await TwoNodeClusterFixture.StartAsync(
            role: "site-int", portA: portHigh, portB: portLow);

        // Real S&F storage + replication actor per node, production default predicate
        // (no isActiveOverride) — the exact wiring under test.
        var (storageOldest, _) = await CreateReplicationActorAsync(fixture.NodeA, "oldest");
        var (storageJoiner, _) = await CreateReplicationActorAsync(fixture.NodeB, "joiner");

        // The delivering (oldest) node has a live buffered row the standby never saw.
        await storageOldest.EnqueueAsync(NewMessage("live-row"));

        // Trigger peer (re)tracking on both sides: each actor got InitialStateAsSnapshot
        // in PreStart, but the enqueue raced it — re-deliver via a fresh MemberUp is not
        // needed; OnPeerTracked already fired on join. The resync exchange is async:
        // wait until the JOINER holds the row (proves the snapshot flowed oldest→joiner,
        // the correct direction). Pre-fix this times out (the joiner, as leader, never
        // requests) AND the oldest node's row is deleted by the stale wipe.
        await AwaitAsync(async () => await storageJoiner.GetMessageByIdAsync("live-row") != null,
            TimeSpan.FromSeconds(20),
            "joiner never received the resync snapshot (resync ran in the wrong direction)");

        // And the delivering node's buffer is untouched — the N1 wipe assertion.
        Assert.NotNull(await storageOldest.GetMessageByIdAsync("live-row"));
    }

    private static async Task<(StoreAndForwardStorage Storage, IActorRef Actor)> CreateReplicationActorAsync(
        ActorSystem node, string tag)
    {
        var sfDb = Path.Combine(Path.GetTempPath(), $"sf-resync-{tag}-{Guid.NewGuid():N}.db");
        var siteDb = Path.Combine(Path.GetTempPath(), $"site-resync-{tag}-{Guid.NewGuid():N}.db");
        var sfStorage = new StoreAndForwardStorage($"Data Source={sfDb}",
            NullLogger<StoreAndForwardStorage>.Instance);
        await sfStorage.InitializeAsync();
        var siteStorage = new SiteStorageService($"Data Source={siteDb}",
            NullLogger<SiteStorageService>.Instance);
        var replicationService = new ReplicationService(
            new StoreAndForwardOptions(), NullLogger<ReplicationService>.Instance);
        // Name MUST be "site-replication" — SendToPeer targets /user/site-replication.
        var actor = node.ActorOf(Props.Create(() => new SiteReplicationActor(
            siteStorage, sfStorage, replicationService, "site-int",
            NullLogger<SiteReplicationActor>.Instance, null, null, null, null)),
            "site-replication");
        return (sfStorage, actor);
    }

    private static StoreAndForwardMessage NewMessage(string id) => new()
    {
        Id = id,
        Category = StoreAndForwardCategory.Notification,
        Target = "central",
        PayloadJson = "{}",
        CreatedAt = DateTimeOffset.UtcNow,
        Status = StoreAndForwardMessageStatus.Pending,
        MaxRetries = 0,
    };

    private static async Task AwaitAsync(Func<Task<bool>> condition, TimeSpan timeout, string why)
    {
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (await condition()) return;
            await Task.Delay(250);
        }
        throw new TimeoutException(why);
    }
}

(Adapt member names to the real SiteStorageService/SiteReplicationActor ctor order — see SiteReplicationActor.cs:64-73 and the harness in tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs:48-70. GetFreeTcpPort may need its access raised to public/internal on the fixture — additive.) 3. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.IntegrationTests --filter SfBufferResyncPredicateTests — expect FAIL on unfixed code (either the timeout — resync ran leader-ward — or the final Assert.NotNull after the wipe). This failure is the deliverable of this task; do NOT fix it here. 4. Commit (the red test rides with the fix task's history; commit it now so Task 3's diff shows red→green): git add tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster && git commit -m "test(saf): failing two-node repro — leader-vs-oldest resync divergence wipes delivering node buffer (plan R2-02 T2)"

Task 3: Swap SiteReplicationActor to the shared oldest-Up predicate + Host wires the delivery-gate delegate

Classification: high-risk (resync authority semantics — the N1 fix) Estimated implement time: ~5 min Parallelizable with: 8, 9, 10, 13 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs (DefaultIsActive lines 190-198; ctor doc lines 56-61; HandleSfBufferSnapshot lines 420-444 belt-and-braces)
• Modify: src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs (RegisterSiteActorsAsync: replication actor creation lines 775-779; provider resolve lines 878-882 hoisted)
• Test: tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs (Task 2's red test goes green), tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs (regression)

Depends on Tasks 1 (evaluator) and 2 (red repro). TDD note (mirrors round-1 Task 4's pattern): this task writes no new test — its failing test is Task 2's repro, plus the existing resync suite as regression.

1. Replace DefaultIsActive (SiteReplicationActor.cs:190-198) and its doc:

    /// <summary>
    /// Repo-standard active-node check: this node is active when it is the OLDEST Up
    /// member carrying the site role — the same oldest-Up semantics as the S&F delivery
    /// gate (IClusterNodeProvider.SelfIsPrimary → ClusterActivityEvaluator → shared
    /// ActiveNodeEvaluator). NEVER the cluster leader: leadership is lowest-address and
    /// diverges from singleton/delivery placement permanently after the lower-address
    /// node restarts — the divergence that made the delivering node wipe its own live
    /// buffer via a wrong-direction resync (review 02 round 2, N1 Critical). Any other
    /// state reports standby — safe-by-default.
    /// </summary>
    private bool DefaultIsActive() =>
        Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(_cluster, _siteRole);

Update the ctor isActiveOverride doc (lines 56-61): the "swap point for plan 01's shared helper" sentence becomes "production wiring passes the Host's IClusterNodeProvider.SelfIsPrimary delegate (the same instance gating the S&F delivery sweep); null falls back to the shared oldest-Up evaluator." 2. Belt-and-braces in HandleSfBufferSnapshot (report's explicit suggestion): re-evaluate the predicate at apply time, inside the async continuation, so an active-flip between receive and apply can never run ReplaceAllAsync on a node that just became the delivering node:

        Task.Run(async () =>
            {
                // Belt-and-braces (N1): re-check at apply time. ReplaceAllAsync discards
                // every in-flight row (StoreAndForwardStorage.cs "Never call on an active
                // node"); a flip between message receipt and this point must abort.
                if (SafeIsActive())
                {
                    _logger.LogWarning(
                        "Discarding S&F buffer resync snapshot: this node became active before apply");
                    return;
                }
                await _sfStorage.ReplaceAllAsync(msg.Messages);
            })
            .ContinueWith(t =>
            {
                if (t.IsFaulted)
                    _logger.LogError(t.Exception, "Failed to apply S&F buffer resync snapshot");
            });

3. Host wiring (AkkaHostedService.RegisterSiteActorsAsync): hoist the clusterNodeProvider resolve (currently at line 878, inside the S&F block) to just above the replication-actor creation at line 775, and pass the SAME delegate:

        // ONE active-node predicate instance governs the S&F delivery gate, the resync
        // authority checks (SiteReplicationActor), and the heartbeat IsActive stamp
        // (SiteCommunicationActor, wired below) — review 02 round 2, N1. Null in
        // non-clustered test hosts: the actors fall back to the shared oldest-Up
        // evaluator, never to a leader check.
        var clusterNodeProvider = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.HealthMonitoring.IClusterNodeProvider>();
        Func<bool>? activeNodeCheck = clusterNodeProvider != null
            ? () => clusterNodeProvider.SelfIsPrimary
            : null;

        var replicationActor = _actorSystem!.ActorOf(
            Props.Create(() => new SiteReplicationActor(
                storage, sfStorage, replicationService, siteRole, replicationLogger,
                deploymentConfigFetcher, activeNodeCheck, siteRuntimeOptionsValue, null)),
            "site-replication");

Then at line ~878 delete the duplicate resolve and keep storeAndForwardService.SetDeliveryGate(() => clusterNodeProvider.SelfIsPrimary); guarded by the hoisted variable. 4. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.IntegrationTests --filter SfBufferResyncPredicateTests — Task 2's repro now PASSES (the oldest node keeps live-row and seeds the joiner). Regression: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter SiteReplicationActorTests (the resync tests inject isActiveOverride, so they are predicate-agnostic — expect PASS) and dotnet test tests/ZB.MOM.WW.ScadaBridge.Host.Tests --filter ClusterActivity. 5. Commit: git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs && git commit -m "fix(saf): resync authority uses the shared oldest-Up predicate + delivery-gate delegate; apply-time re-check guard (plan R2-02 T3)"

Task 4: Swap SiteCommunicationActor.DefaultIsActiveCheck + Host wiring + doc sync

Classification: small (predicate swap + wiring; divergence coverage is Tasks 1-3) Estimated implement time: ~4 min Parallelizable with: 8, 9, 10, 13, 15 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs (DefaultIsActiveCheck lines 508-526; ctor doc lines 68-74)
• Modify: src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs (siteCommActor creation lines 822-827 — pass activeNodeCheck)
• Modify: docs/requirements/Component-Communication.md (any "leader" wording on the heartbeat IsActive stamp), docs/requirements/Component-StoreAndForward.md line 83 ("repo-standard leader+Up active-node predicate" → oldest-Up; the fuller resync rewrite lands in Task 7)

Depends on Tasks 1 and 3 (shared AkkaHostedService.cs — serialized).

1. Replace DefaultIsActiveCheck (SiteCommunicationActor.cs:517-526):

    /// <summary>
    /// Default active-node check used when no override is supplied: oldest-Up member
    /// semantics via the shared <see cref="ClusterState.ActiveNodeEvaluator"/> — the
    /// same predicate as the S&F delivery gate and the replication resync authority
    /// (review 02 round 2, N1). Unscoped by role: a site cluster's members all carry
    /// the site role, so role scoping is a no-op here; production wiring passes the
    /// Host's role-scoped IClusterNodeProvider delegate anyway. Any other state
    /// (still joining, leaving) reports standby — safe-by-default.
    /// </summary>
    private bool DefaultIsActiveCheck() =>
        ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(Cluster.Get(Context.System));

Update the ctor doc (lines 68-74): "…null uses the shared oldest-Up evaluator (production wiring passes the Host's singleton-host delegate)…". 2. Host wiring (line 822-827) — pass the Task 3 delegate as the 4th ctor arg:

        var siteCommActor = _actorSystem.ActorOf(
            Props.Create(() => new SiteCommunicationActor(
                _nodeOptions.SiteId!,
                _communicationOptions,
                dmProxy,
                activeNodeCheck)),
            "site-communication");

3. Docs: grep both component docs for leader in the heartbeat/resync context and align to "oldest-Up member (singleton-host semantics, shared ActiveNodeEvaluator)". Do NOT rewrite the full resync paragraph yet (Task 7 owns it — avoid a double edit).
4. Build + regression: dotnet build ZB.MOM.WW.ScadaBridge.slnx, then dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter SiteCommunicationActor (existing tests inject isActiveCheck stubs — expect PASS).
5. Commit: git add src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs docs/requirements/Component-Communication.md docs/requirements/Component-StoreAndForward.md && git commit -m "fix(comm): heartbeat IsActive uses the shared oldest-Up predicate; Host wires one delegate everywhere (plan R2-02 T4)"

Task 5: Chunked resync protocol — additive messages, byte-budgeted chunker, active-side chunked answer

Classification: high-risk (cluster protocol; frame-size correctness — the N2 fix, part 1) Estimated implement time: ~5 min Parallelizable with: 8, 9, 10, 13, 14, 15 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs (new records; HandleRequestSfBufferResync lines 398-411; new chunker + constants near MaxResyncRows line 40)
• Test: tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs (append), tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs (create)

Depends on Task 3 (same file). Message-contract rules (additive-only): SfBufferSnapshotChunk and SfBufferResyncAck are NEW records; the legacy SfBufferSnapshot record AND its standby handler are retained so a rolling upgrade keeps working (old active → monolithic snapshot to new standby: handled; new active → chunks to old standby: dead-lettered, divergence self-heals at the next post-upgrade peer-track — same exposure as today's silently-dropped oversized frame, now bounded to the upgrade window and logged). These messages ride intra-site Akka remoting, not ClusterClient, so ClusterClientContractLockTests (PLAN-08 T9, tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Messages/ClusterClientContractLockTests.cs) is deliberately untouched — the wire pins land in a new ResyncWireSerializationPinTests following the ReplicationWireSerializationPinTests pattern instead.

Sizing rationale: Akka remoting's default maximum-frame-size is 128 000 bytes and BuildHocon does not raise it. Budget MaxResyncChunkBytes = 64_000 estimated payload per chunk (≈50% headroom for the JSON envelope, type manifests, and non-payload columns) with a MaxResyncChunkRows = 200 row cap. A single row whose payload alone exceeds the budget ships as its own chunk with a Warning (it may still exceed the frame — pathological, logged, no longer silent for everything else). The active node still materializes at most MaxResyncRows (10 000) rows in memory — unchanged from today and bounded; accepted (report: "reconsider" — the cap already bounds it).

1. Failing tests (append to SiteReplicationActorTests.cs; reuse NewMessage/ResyncTestActor helpers):

    // ── R2 T5: chunked resync answer ──

    [Fact]
    public void ChunkForRemoting_SplitsByByteBudget_PreservingOrderAndSequence()
    {
        var rows = Enumerable.Range(0, 10)
            .Select(i => NewMessage($"m{i}", payloadJson: new string('x', 20_000)))
            .ToList();

        var chunks = SiteReplicationActor.ChunkForRemoting(rows, maxChunkBytes: 64_000, maxChunkRows: 200);

        Assert.True(chunks.Count > 1); // 10 × 20 KB cannot ride one 64 KB chunk
        Assert.Equal(rows.Select(r => r.Id), chunks.SelectMany(c => c).Select(r => r.Id)); // order preserved
        Assert.All(chunks, c => Assert.True(
            c.Sum(r => r.PayloadJson.Length) <= 64_000 || c.Count == 1)); // budget honored (oversized row isolated)
    }

    [Fact]
    public void ChunkForRemoting_RowCapHonored_AndSingleOversizedRowIsolated()
    {
        var many = Enumerable.Range(0, 500).Select(i => NewMessage($"s{i}", payloadJson: "{}")).ToList();
        Assert.All(SiteReplicationActor.ChunkForRemoting(many, 64_000, 200), c => Assert.True(c.Count <= 200));

        var oversized = new List<StoreAndForwardMessage>
            { NewMessage("big", payloadJson: new string('y', 100_000)), NewMessage("small", payloadJson: "{}") };
        var chunks = SiteReplicationActor.ChunkForRemoting(oversized, 64_000, 200);
        Assert.Equal(2, chunks.Count); // the oversized row rides alone
    }

    [Fact]
    public async Task ActiveNode_AnswersResyncRequest_WithSequencedChunks_SharingOneResyncId()
    {
        for (var i = 0; i < 3; i++)
            await _sfStorage.EnqueueAsync(NewMessage($"c{i}", payloadJson: new string('z', 30_000)));
        var actor = CreateResyncActor(isActive: () => true);

        actor.Tell(new RequestSfBufferResync(), TestActor);

        var first = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
        var rest = Enumerable.Range(1, first.TotalChunks - 1)
            .Select(_ => ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5)))
            .Prepend(first)
            .ToList();

        Assert.True(first.TotalChunks > 1);
        Assert.All(rest, c => Assert.Equal(first.ResyncId, c.ResyncId));
        Assert.Equal(Enumerable.Range(1, first.TotalChunks), rest.Select(c => c.Sequence));
        Assert.Equal(3, rest.Sum(c => c.Messages.Count));
    }

(NewMessage may need an optional payloadJson parameter — additive to the existing helper. CreateResyncActor = the existing ResyncTestActor construction.) 2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "ChunkForRemoting|WithSequencedChunks" — expect FAIL (compile: chunker/records missing). 3. Implement in SiteReplicationActor.cs:

• Constants next to MaxResyncRows (line 40): internal const int MaxResyncChunkBytes = 64_000; and internal const int MaxResyncChunkRows = 200; with the sizing-rationale doc above (cite the 128 000-byte default frame and that BuildHocon sets no override).
• Pure chunker (internal static, unit-testable):

    /// <summary>
    /// Splits a resync snapshot into chunks that fit Akka remoting's default
    /// 128 000-byte frame (review 02 round 2, N2): rows accumulate until the estimated
    /// payload budget or the row cap is hit. Estimation is payload-dominated
    /// (payload_json length + 512 bytes fixed overhead per row); a single row whose
    /// payload exceeds the budget ships alone (Warning at the call site). Order is
    /// preserved (oldest-first, matching GetAllMessagesAsync).
    /// </summary>
    internal static List<List<StoreAndForwardMessage>> ChunkForRemoting(
        IReadOnlyList<StoreAndForwardMessage> rows, int maxChunkBytes, int maxChunkRows)
    {
        var chunks = new List<List<StoreAndForwardMessage>>();
        var current = new List<StoreAndForwardMessage>();
        var currentBytes = 0;
        foreach (var row in rows)
        {
            var estimate = (row.PayloadJson?.Length ?? 0) + 512;
            if (current.Count > 0 && (currentBytes + estimate > maxChunkBytes || current.Count >= maxChunkRows))
            {
                chunks.Add(current);
                current = new List<StoreAndForwardMessage>();
                currentBytes = 0;
            }
            current.Add(row);
            currentBytes += estimate;
        }
        if (current.Count > 0) chunks.Add(current);
        return chunks;
    }

• HandleRequestSfBufferResync (lines 398-411): keep the authority check; pipe the snapshot back to Self (chunking on the actor thread keeps SendToPeer/ack bookkeeping actor-safe), then send sequenced chunks to the requester:

        var replyTo = Sender;
        _sfStorage.GetAllMessagesAsync(MaxResyncRows).PipeTo(
            Self,
            failure: ex => new Status.Failure(ex),
            success: result => new SfResyncSnapshotLoaded(replyTo, result.Messages, result.Truncated));

with a new internal record + handler:

    /// <summary>Internal: the resync snapshot finished loading; chunk and send to the requester.</summary>
    internal sealed record SfResyncSnapshotLoaded(
        IActorRef ReplyTo, List<StoreAndForwardMessage> Messages, bool Truncated);

    private void HandleSfResyncSnapshotLoaded(SfResyncSnapshotLoaded msg)
    {
        var chunks = ChunkForRemoting(msg.Messages, MaxResyncChunkBytes, MaxResyncChunkRows);
        if (chunks.Count == 0) chunks.Add(new List<StoreAndForwardMessage>()); // empty buffer still resyncs (clears the standby)
        var resyncId = Guid.NewGuid().ToString("N");
        for (var i = 0; i < chunks.Count; i++)
        {
            if (chunks[i].Count == 1 && (chunks[i][0].PayloadJson?.Length ?? 0) + 512 > MaxResyncChunkBytes)
                _logger.LogWarning(
                    "Resync row {Id} alone exceeds the chunk budget ({Bytes}B payload); sending solo — it may exceed the remoting frame",
                    chunks[i][0].Id, chunks[i][0].PayloadJson?.Length ?? 0);
            msg.ReplyTo.Tell(new SfBufferSnapshotChunk(resyncId, i + 1, chunks.Count, chunks[i], msg.Truncated), Self);
        }
        _logger.LogInformation(
            "Answered S&F resync request with {Rows} row(s) in {Chunks} chunk(s), resyncId={ResyncId}",
            msg.Messages.Count, chunks.Count, resyncId);
        // Task 7 registers the pending-ack entry here.
    }

• New public records at the bottom of the file (next to SfBufferSnapshot, which is KEPT verbatim for rolling compat):

/// <summary>
/// Active→standby: one sequenced chunk of a full-buffer anti-entropy snapshot
/// (review 02 round 2, N2 — the monolithic <see cref="SfBufferSnapshot"/> exceeds Akka
/// remoting's default 128 000-byte frame for any realistic backlog). All chunks of one
/// resync share <paramref name="ResyncId"/>; <paramref name="Sequence"/> is 1-based up to
/// <paramref name="TotalChunks"/>. Additive message — the legacy monolithic snapshot
/// handler is retained for rolling upgrades. Crosses intra-site Akka remoting (NOT
/// ClusterClient — ClusterClientContractLockTests is intentionally not involved).
/// </summary>
public sealed record SfBufferSnapshotChunk(
    string ResyncId, int Sequence, int TotalChunks,
    List<StoreAndForwardMessage> Messages, bool Truncated);

/// <summary>
/// Standby→active: delivery confirmation — the standby assembled all chunks of
/// <paramref name="ResyncId"/> and applied them atomically (<paramref name="RowCount"/>
/// rows installed). Absence within the ack window is surfaced by the active node
/// (Warning + counter) — the silent-loss mode N2 flagged.
/// </summary>
public sealed record SfBufferResyncAck(string ResyncId, int RowCount);

• Register Receive<SfResyncSnapshotLoaded>(HandleSfResyncSnapshotLoaded); in the ctor next to the other resync receives (line 111-113).

4. Create tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs following tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationWireSerializationPinTests.cs: round-trip SfBufferSnapshotChunk (fully-populated StoreAndForwardMessage incl. ExecutionId/ParentExecutionId) and SfBufferResyncAck through Sys.Serialization, plus type-identity pins (ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.SfBufferSnapshotChunk, ...SfBufferResyncAck) — a rename/move is an intra-site wire break during rolling upgrades.
5. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "ChunkForRemoting|WithSequencedChunks|ResyncWireSerializationPin" — expect PASS. Then the full resync group: --filter SiteReplicationActorTests.
6. Commit: git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests && git commit -m "fix(saf): chunk the resync snapshot to fit Akka remoting frames (additive protocol, byte-budgeted) (plan R2-02 T5)"

Task 6: Standby-side chunk assembly, atomic apply, ack + the N5 race comment

Classification: high-risk (cluster protocol; buffer-replace correctness — the N2 fix, part 2) Estimated implement time: ~5 min Parallelizable with: 8, 9, 10, 13, 14, 15 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs (new assembly state + Receive<SfBufferSnapshotChunk>; N5 doc comment on the apply path)
• Test: tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs (append)

Depends on Task 5 (same file, chunk records).

1. Failing tests:

    [Fact]
    public async Task StandbyNode_AssemblesChunks_AppliesOnce_AndAcks()
    {
        await _sfStorage.EnqueueAsync(NewMessage("stale"));
        var actor = CreateResyncActor(isActive: () => false);
        var resyncId = "r1";

        actor.Tell(new SfBufferSnapshotChunk(resyncId, 1, 2,
            new List<StoreAndForwardMessage> { NewMessage("f1") }, false), TestActor);
        actor.Tell(new SfBufferSnapshotChunk(resyncId, 2, 2,
            new List<StoreAndForwardMessage> { NewMessage("f2") }, false), TestActor);

        var ack = ExpectMsg<SfBufferResyncAck>(TimeSpan.FromSeconds(5));
        Assert.Equal(resyncId, ack.ResyncId);
        Assert.Equal(2, ack.RowCount);
        await AwaitAssertAsync(async () =>
        {
            Assert.Null(await _sfStorage.GetMessageByIdAsync("stale"));   // replaced wholesale
            Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f1"));
            Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f2"));
        });
    }

    [Fact]
    public async Task StandbyNode_NewResyncId_DiscardsStalePartialAssembly()
    {
        var actor = CreateResyncActor(isActive: () => false);
        actor.Tell(new SfBufferSnapshotChunk("old", 1, 2,
            new List<StoreAndForwardMessage> { NewMessage("orphan") }, false), TestActor);
        actor.Tell(new SfBufferSnapshotChunk("new", 1, 1,
            new List<StoreAndForwardMessage> { NewMessage("fresh") }, false), TestActor);

        ExpectMsg<SfBufferResyncAck>(TimeSpan.FromSeconds(5)); // "new" completed
        await AwaitAssertAsync(async () =>
        {
            Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh"));
            Assert.Null(await _sfStorage.GetMessageByIdAsync("orphan")); // stale partial never applied
        });
    }

    [Fact]
    public void ActiveNode_IgnoresChunks_NeverAcks()
    {
        var actor = CreateResyncActor(isActive: () => true);
        actor.Tell(new SfBufferSnapshotChunk("r", 1, 1,
            new List<StoreAndForwardMessage> { NewMessage("x") }, false), TestActor);
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
    }

2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "AssemblesChunks|DiscardsStalePartialAssembly|IgnoresChunks_NeverAcks" — expect FAIL (no chunk handler; chunks dead-letter).
3. Implement — actor gains IWithTimers (public ITimerScheduler Timers { get; set; } = null!;) plus assembly state and the handler:

    // ── Chunked-resync assembly (standby side; actor-thread only) ──
    private string? _assemblingResyncId;
    private int _assemblingTotalChunks;
    private bool _assemblingTruncated;
    private readonly Dictionary<int, List<StoreAndForwardMessage>> _assemblingChunks = new();
    private const string ResyncAssemblyTimerKey = "sf-resync-assembly-timeout";

    /// <summary>How long a partial chunk assembly may wait for its missing chunks before
    /// being discarded (a lost chunk = lost resync; the next peer-track retries). Ctor
    /// test seam; production default 30 s.</summary>
    private readonly TimeSpan _resyncAssemblyTimeout;

(ctor gains TimeSpan? resyncAssemblyTimeout = null → _resyncAssemblyTimeout = resyncAssemblyTimeout ?? TimeSpan.FromSeconds(30); — additive optional parameter, existing callers unchanged.)

    private void HandleSfBufferSnapshotChunk(SfBufferSnapshotChunk msg)
    {
        if (SafeIsActive())
        {
            _logger.LogDebug("Ignoring S&F resync chunk — this node is active");
            return;
        }

        if (_assemblingResyncId != msg.ResyncId)
        {
            if (_assemblingResyncId != null)
                _logger.LogWarning(
                    "Discarding partial S&F resync assembly {Old} ({Have}/{Want} chunks): a new resync {New} superseded it",
                    _assemblingResyncId, _assemblingChunks.Count, _assemblingTotalChunks, msg.ResyncId);
            _assemblingResyncId = msg.ResyncId;
            _assemblingTotalChunks = msg.TotalChunks;
            _assemblingTruncated = msg.Truncated;
            _assemblingChunks.Clear();
        }

        _assemblingChunks[msg.Sequence] = msg.Messages;
        Timers.StartSingleTimer(ResyncAssemblyTimerKey, new ResyncAssemblyTimedOut(msg.ResyncId), _resyncAssemblyTimeout);

        if (_assemblingChunks.Count < _assemblingTotalChunks)
            return;

        // Complete: assemble in sequence order and apply atomically.
        var assembled = Enumerable.Range(1, _assemblingTotalChunks)
            .SelectMany(seq => _assemblingChunks[seq])
            .ToList();
        var resyncId = _assemblingResyncId!;
        var truncated = _assemblingTruncated;
        _assemblingResyncId = null;
        _assemblingChunks.Clear();
        Timers.Cancel(ResyncAssemblyTimerKey);

        if (truncated)
            _logger.LogWarning(
                "S&F buffer resync snapshot truncated at {Cap} rows; divergence beyond the cap drains naturally",
                MaxResyncRows);
        _logger.LogInformation(
            "Applying chunked S&F resync {ResyncId} ({Count} rows), replacing local buffer", resyncId, assembled.Count);

        // KNOWN, ACCEPTED race (review 02 round 2, N5 — do NOT "fix" this into something
        // worse): a replicated Remove sent after the active node read its snapshot but
        // before the snapshot's chunks is ordered BEFORE them on the wire (same
        // sender/receiver pair), so this apply can re-add the removed row → an orphan
        // Pending row that a later failover re-delivers ONCE. Bounded, self-correcting at
        // the next resync, and inherent to no-ack replication; a delivered-side dedup or
        // op-sequencing scheme would cost far more than one rare duplicate.
        var replyTo = Sender;
        Task.Run(async () =>
            {
                if (SafeIsActive()) // belt-and-braces, mirrors the monolithic path (T3)
                {
                    _logger.LogWarning("Discarding chunked S&F resync {ResyncId}: node became active before apply", resyncId);
                    return;
                }
                await _sfStorage.ReplaceAllAsync(assembled);
                replyTo.Tell(new SfBufferResyncAck(resyncId, assembled.Count));
            })
            .ContinueWith(t =>
            {
                if (t.IsFaulted)
                    _logger.LogError(t.Exception, "Failed to apply chunked S&F resync {ResyncId}", resyncId);
            });
    }

    private void HandleResyncAssemblyTimedOut(ResyncAssemblyTimedOut msg)
    {
        if (_assemblingResyncId != msg.ResyncId) return; // superseded already
        _logger.LogWarning(
            "S&F resync assembly {ResyncId} timed out with {Have}/{Want} chunks — discarding partial (a lost chunk; the next peer-track retries)",
            msg.ResyncId, _assemblingChunks.Count, _assemblingTotalChunks);
        ScadaBridgeTelemetry.RecordReplicationFailure();
        _assemblingResyncId = null;
        _assemblingChunks.Clear();
    }

Plus internal sealed record ResyncAssemblyTimedOut(string ResyncId); and ctor registrations Receive<SfBufferSnapshotChunk>(HandleSfBufferSnapshotChunk); / Receive<ResyncAssemblyTimedOut>(HandleResyncAssemblyTimedOut); next to the legacy Receive<SfBufferSnapshot> (line 113 — which stays, for rolling compat). Sender inside HandleSfBufferSnapshotChunk is the active node's replication actor (chunks are Tell'd with Self as sender in T5), so the ack lands where Task 7 listens. 4. Run the filter — expect PASS. Full group: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter SiteReplicationActorTests, and re-run the Task 2 integration repro (--filter SfBufferResyncPredicateTests) — it must still pass with the exchange now chunked. 5. Commit: git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs && git commit -m "fix(saf): standby assembles chunked resync, applies atomically with re-check, acks; N5 race documented (plan R2-02 T6)"

Task 7: Resync delivery confirmation on the active node + telemetry + doc rewrite

Classification: standard Estimated implement time: ~5 min Parallelizable with: 8, 9, 10, 13, 15 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs (ack tracking in HandleSfResyncSnapshotLoaded + Receive<SfBufferResyncAck>), src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs (two counters, next to RecordReplicationFailure at line 108)
• Modify: docs/requirements/Component-StoreAndForward.md (line 83 — full resync-paragraph rewrite)
• Test: tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs (append)

Depends on Task 6 (same file). N2's "no telemetry counts it / nothing retries until the next peer-track" — the ack closes the silent-loss mode: a resync that was answered but never applied is now a Warning + counter instead of nothing.

1. Failing tests:

    [Fact]
    public async Task ActiveNode_ReceivingAck_CountsResyncCompleted()
    {
        await _sfStorage.EnqueueAsync(NewMessage("m1"));
        var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromSeconds(30));
        actor.Tell(new RequestSfBufferResync(), TestActor);
        var chunk = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));

        actor.Tell(new SfBufferResyncAck(chunk.ResyncId, 1), TestActor);
        // Observable: the ack-timeout no longer fires (next test proves the inverse);
        // completing must not warn. EventFilter pins it:
        // (arrange the EventFilter *around* the ack Tell)
    }

    [Fact]
    public async Task ActiveNode_MissingAck_WarnsAfterAckTimeout()
    {
        await _sfStorage.EnqueueAsync(NewMessage("m1"));
        var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromMilliseconds(200));

        await EventFilter.Warning(contains: "was never acknowledged").ExpectOneAsync(async () =>
        {
            actor.Tell(new RequestSfBufferResync(), TestActor);
            ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
            await Task.Delay(500); // let the ack timer fire
        });
    }

(CreateResyncActor gains an ackTimeout pass-through to the new ctor seam.) 2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "CountsResyncCompleted|WarnsAfterAckTimeout" — expect FAIL (no ack handling; ctor param missing). 3. Implement:

• ScadaBridgeTelemetry (alongside RecordReplicationFailure, line 108):

    private static readonly Counter<long> _sfResyncCompleted =
        Meter.CreateCounter<long>("scadabridge.store_and_forward.resync.completed", unit: "1",
            description: "S&F anti-entropy resyncs the standby acknowledged as applied");
    /// <summary>Records one acknowledged (applied) S&F buffer resync.</summary>
    public static void RecordSfResyncCompleted() => _sfResyncCompleted.Add(1);

    private static readonly Counter<long> _sfResyncAckMissing =
        Meter.CreateCounter<long>("scadabridge.store_and_forward.resync.ack_missing", unit: "1",
            description: "S&F resyncs answered by the active node but never acknowledged by the standby within the ack window (lost chunks / dead peer)");
    /// <summary>Records one resync whose ack window expired.</summary>
    public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.Add(1);

• SiteReplicationActor: ctor seam TimeSpan? resyncAckTimeout = null → _resyncAckTimeout (default 60 s); in HandleSfResyncSnapshotLoaded after the chunk loop: _pendingAckResyncId = resyncId; Timers.StartSingleTimer(ResyncAckTimerKey, new ResyncAckTimedOut(resyncId), _resyncAckTimeout); plus:

        Receive<SfBufferResyncAck>(msg =>
        {
            if (msg.ResyncId == _pendingAckResyncId)
            {
                _pendingAckResyncId = null;
                Timers.Cancel(ResyncAckTimerKey);
            }
            ScadaBridgeTelemetry.RecordSfResyncCompleted();
            _logger.LogInformation("S&F resync {ResyncId} acknowledged by standby: {Rows} row(s) applied",
                msg.ResyncId, msg.RowCount);
        });
        Receive<ResyncAckTimedOut>(msg =>
        {
            if (msg.ResyncId != _pendingAckResyncId) return;
            _pendingAckResyncId = null;
            ScadaBridgeTelemetry.RecordSfResyncAckMissing();
            _logger.LogWarning(
                "S&F resync {ResyncId} was never acknowledged within {Window} — snapshot chunks may have been lost (frame drop / dead peer); the next peer-track retries",
                msg.ResyncId, _resyncAckTimeout);
        });

(single-outstanding-resync bookkeeping is sufficient: a new request supersedes — overwrite _pendingAckResyncId and restart the timer.)

• Doc rewrite — Component-StoreAndForward.md:83, replace the resync paragraph: chunked SfBufferSnapshotChunk protocol (byte-budgeted for Akka remoting's 128 KB default frame), standby assembly + atomic ReplaceAllAsync + SfBufferResyncAck confirmation, ack-missing Warning + counters, the shared oldest-Up predicate on both roles (same predicate as the delivery gate — cite N1), the retained legacy monolithic handler for rolling upgrades, and the accepted N5 orphan-row race.

4. Run the filter + --filter SiteReplicationActorTests — expect PASS.
5. Commit: git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs docs/requirements/Component-StoreAndForward.md tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs && git commit -m "feat(saf): resync ack confirmation + completed/ack-missing telemetry; doc resync rewrite (plan R2-02 T7)"

Task 8: Publish _sweepTask only when the sweep CAS is won (unclobber the shutdown drain)

Classification: high-risk (shutdown/concurrency semantics) Estimated implement time: ~5 min Parallelizable with: 1, 2, 3, 4, 5, 6, 7, 9, 10, 13, 14, 15 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs (timer callback line 396-400; TriggerSweep lines 664-668; RetryPendingMessagesAsync lines 674-744 split)
• Test: tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs (append)

N3: both the timer callback and TriggerSweep unconditionally Volatile.Write(ref _sweepTask, RetryPendingMessagesAsync()); while a sweep is in flight the new call no-ops via the _retryInProgress CAS and returns a completed task that overwrites the drain handle — StopAsync (line 422-431) then awaits the no-op and shuts down under the real sweep. Fix: the caller performs the CAS and publishes the task only on a win.

1. Failing tests (append; CreateService(retryTimerInterval) helper exists at line 53 of the test file):

    [Fact]
    public async Task TriggerSweep_WhileSweepInFlight_DoesNotClobberTheDrainHandle()
    {
        var service = CreateService(retryTimerInterval: TimeSpan.FromHours(1));
        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, async _ =>
        {
            entered.TrySetResult();
            await release.Task;
            return true;
        });
        await service.StartAsync();
        try
        {
            await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}",
                attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);

            service.TriggerSweep();                       // real sweep, blocked in the handler
            await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
            service.TriggerSweep();                       // redundant kick — pre-fix clobbers _sweepTask

            var handle = service.CurrentSweepTaskForTest;
            Assert.NotNull(handle);
            Assert.False(handle!.IsCompleted);            // pre-fix: true (a completed no-op replaced the real sweep)
        }
        finally
        {
            release.TrySetResult();
            await service.StopAsync();
        }
    }

    [Fact]
    public async Task StopAsync_WaitsForTheRealInFlightSweep_EvenAfterARedundantTrigger()
    {
        var service = CreateService(retryTimerInterval: TimeSpan.FromHours(1));
        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, async _ =>
        {
            entered.TrySetResult();
            await release.Task;
            return true;
        });
        await service.StartAsync();
        await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}",
            attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);

        service.TriggerSweep();
        await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
        service.TriggerSweep(); // the clobbering kick

        var stop = service.StopAsync();
        await Task.Delay(300);
        Assert.False(stop.IsCompleted); // pre-fix: StopAsync already returned (awaited the no-op)

        release.TrySetResult();
        await stop.WaitAsync(TimeSpan.FromSeconds(5)); // drains the real sweep promptly once released
    }

2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "DoesNotClobberTheDrainHandle|WaitsForTheRealInFlightSweep" — expect FAIL (CurrentSweepTaskForTest missing; then the pre-fix clobber assertions).
3. Implement in StoreAndForwardService:

    /// <summary>The current drain handle — test seam for the N3 clobber regression.</summary>
    internal Task? CurrentSweepTaskForTest => Volatile.Read(ref _sweepTask);

    /// <summary>
    /// Starts a sweep IF none is in flight, publishing the new task into
    /// <see cref="_sweepTask"/> only when this call wins the <see cref="_retryInProgress"/>
    /// CAS. A kick that loses the CAS returns without touching <see cref="_sweepTask"/> —
    /// pre-fix it overwrote the drain handle with an instantly-completed no-op, so
    /// <see cref="StopAsync"/> proceeded with disposal under a still-running sweep
    /// (review 02 round 2, N3) — defeating the drain exactly when a sweep outlives a tick.
    /// </summary>
    private void KickSweep()
    {
        if (Interlocked.CompareExchange(ref _retryInProgress, 1, 0) != 0)
            return;
        Volatile.Write(ref _sweepTask, RunSweepOwnedAsync());
    }

• RetryPendingMessagesAsync keeps its signature for the existing direct-call tests, but becomes the CAS wrapper: internal async Task RetryPendingMessagesAsync() { if (Interlocked.CompareExchange(ref _retryInProgress, 1, 0) != 0) return; await RunSweepOwnedAsync(); } — and the existing body (lines 680-743, including the delivery gate, lanes, catch, and the finally { Interlocked.Exchange(ref _retryInProgress, 0); }) moves verbatim into private async Task RunSweepOwnedAsync() with the entry CAS removed (ownership is the caller's).
• Timer callback (line 397): _ => KickSweep().
• TriggerSweep (line 664-668): if (_retryTimer == null) return; KickSweep(); (doc comment updated: "publishes into _sweepTask only when the CAS is won — see KickSweep").

4. Run the filter, then the whole project's sweep group: dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "RetrySweep_|Enqueue_DeferToSweep|StopAsync" — expect PASS (existing sweep/defer tests unchanged: RetryPendingMessagesAsync still self-CASes for direct calls).
5. Commit: git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs && git commit -m "fix(saf): publish _sweepTask only on CAS win so StopAsync drains the real in-flight sweep (plan R2-02 T8)"

Task 9: Eager validation for SweepBatchLimit / SweepTargetParallelism

Classification: trivial Estimated implement time: ~3 min Parallelizable with: 1, 2, 3, 4, 5, 6, 7, 8, 10, 11, 13, 14, 15 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs (lines 17-34)
• Test: tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs (append)

N4: a negative SweepBatchLimit silently means "unlimited (legacy)" (StoreAndForwardStorage.cs:393 checks > 0); a non-positive SweepTargetParallelism is silently clamped to 1 (StoreAndForwardService.cs:715). Both get fail-fast messages per the §1.5 eager-validation convention (PLAN-08).

1. Failing tests (append, following the file's existing valid/invalid pattern):

    [Theory]
    [InlineData(-1)]
    [InlineData(-500)]
    public void Validate_NegativeSweepBatchLimit_Fails(int limit)
    {
        var options = ValidOptions();
        options.SweepBatchLimit = limit;
        var result = Validate(options);
        Assert.True(result.Failed);
        Assert.Contains("SweepBatchLimit", result.FailureMessage);
    }

    [Fact]
    public void Validate_ZeroSweepBatchLimit_IsValidUnlimitedLegacy()
    {
        var options = ValidOptions();
        options.SweepBatchLimit = 0;
        Assert.True(Validate(options).Succeeded);
    }

    [Theory]
    [InlineData(0)]
    [InlineData(-4)]
    public void Validate_NonPositiveSweepTargetParallelism_Fails(int parallelism)
    {
        var options = ValidOptions();
        options.SweepTargetParallelism = parallelism;
        var result = Validate(options);
        Assert.True(result.Failed);
        Assert.Contains("SweepTargetParallelism", result.FailureMessage);
    }

2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "SweepBatchLimit_Fails|ZeroSweepBatchLimit|SweepTargetParallelism_Fails" — expect FAIL (validator passes everything).
3. Implement (append inside Validate, after the DefaultMaxRetries rule at line 31-33):

        builder.RequireThat(options.SweepBatchLimit >= 0,
            $"ScadaBridge:StoreAndForward:SweepBatchLimit must be >= 0 " +
            $"(was {options.SweepBatchLimit}); it bounds due rows per retry sweep — 0 means unlimited (legacy).");

        builder.RequireThat(options.SweepTargetParallelism >= 1,
            $"ScadaBridge:StoreAndForward:SweepTargetParallelism must be >= 1 " +
            $"(was {options.SweepTargetParallelism}); it caps concurrent (category,target) sweep lanes — 1 means serial.");

4. Run the filter + dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter StoreAndForwardOptionsValidator — expect PASS.
5. Commit: git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs && git commit -m "chore(saf): eagerly validate SweepBatchLimit/SweepTargetParallelism (plan R2-02 T9)"

Task 10: Coalesce live-alarm delta publishes (bound the per-circuit fan-out)

Classification: standard (perf; correctness unchanged — last write wins). MUTEX with PLAN-R2-07's sticky-IsLive task — see the rule at the top; do not start while R2-07's live-alarm slice is in flight. Estimated implement time: ~5 min Parallelizable with: 1, 2, 3, 4, 5, 6, 7, 8, 9, 13, 14 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs (HandleLiveDelta lines 347-368; OnSeedCompleted line 303; ctor + new timer), src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs (after line 124), src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs (after line 87), src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs (Props.Create lines 228-236 — pass the new option)
• Test: tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs (append), tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs (append)

N6: HandleLiveDelta → Publish() materializes _cache.Values.ToList() per delta and OnPublish fans out to every viewer circuit per delta — O(cacheSize) copies and a render per circuit per transition during an alarm storm. Fix: dirty-flag + single-shot coalescing timer (default 250 ms; TimeSpan.Zero restores per-delta publish). Seed/reconcile completion still publishes immediately (already batched).

1. Failing tests (append; reuse SeedStub/PublishSink/MockSiteAlarmStreamClient and the existing CreateAggregator-style construction):

    [Fact]
    public void DeltaBurst_IsCoalesced_IntoFewPublishes_ContainingEveryRow()
    {
        var (actor, seed, sink, client) = CreateAggregator(publishCoalesce: TimeSpan.FromMilliseconds(150));
        seed.CompleteNext(); // empty initial seed
        AwaitAssert(() => Assert.Equal(1, sink.Count)); // seed publish, immediate

        var sub = client.Subs.Single();
        for (var i = 0; i < 50; i++)
            sub.OnAlarm(Alarm($"A{i}", DateTimeOffset.UtcNow)); // 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)
        });
    }

    [Fact]
    public void ZeroCoalesce_PreservesLegacyPerDeltaPublish()
    {
        var (actor, seed, sink, client) = CreateAggregator(publishCoalesce: TimeSpan.Zero);
        seed.CompleteNext();
        AwaitAssert(() => Assert.Equal(1, sink.Count));

        var sub = client.Subs.Single();
        sub.OnAlarm(Alarm("A1", DateTimeOffset.UtcNow));
        sub.OnAlarm(Alarm("A2", DateTimeOffset.UtcNow));
        AwaitAssert(() => Assert.Equal(3, sink.Count)); // 1 seed + 1 per delta
    }

2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter "DeltaBurst_IsCoalesced|ZeroCoalesce_Preserves" — expect FAIL (ctor lacks the parameter; then 51 publishes).
3. Implement:
   • CommunicationOptions (after LiveAlarmCacheMaxSubscribersPerSite, line 124):

    /// <summary>
    /// Publish-coalescing window for live alarm deltas: an applied delta marks the cache
    /// dirty and one publish (fresh snapshot + per-viewer onChanged fan-out) fires after
    /// this window, batching an alarm storm into ~4 publishes/second instead of one per
    /// transition (review 02 round 2, N6). Zero = publish per delta (legacy). Seed and
    /// reconcile publishes are always immediate. Default 250 ms.
    /// </summary>
    public TimeSpan LiveAlarmCachePublishCoalesce { get; set; } = TimeSpan.FromMilliseconds(250);

• CommunicationOptionsValidator (after line 87): builder.RequireThat(options.LiveAlarmCachePublishCoalesce >= TimeSpan.Zero, $"Communication:LiveAlarmCachePublishCoalesce must be zero or a positive duration (was {options.LiveAlarmCachePublishCoalesce})."); — plus a validator test in CommunicationOptionsValidatorTests (negative fails, zero succeeds).
• SiteAlarmAggregatorActor: ctor gains TimeSpan publishCoalesce (expression trees reject optional args — every Props.Create/test call site passes it explicitly); fields private readonly TimeSpan _publishCoalesce; private bool _publishPending; and private const string PublishTimerKey = "alarm-publish-coalesce";; HandleLiveDelta's pass-through branch (line 366-367) becomes if (ApplyDelta(delta, requireStrictlyNewer: false)) SchedulePublish(); with:

    /// <summary>
    /// Coalesced publish: with a positive window, the first dirtying delta arms a
    /// single-shot timer and further deltas ride the same tick — one snapshot copy and
    /// one viewer fan-out per window instead of per transition (N6). Zero = legacy
    /// immediate publish. Last write wins, so batching never changes final state.
    /// </summary>
    private void SchedulePublish()
    {
        if (_publishCoalesce <= TimeSpan.Zero) { Publish(); return; }
        if (_publishPending) return;
        _publishPending = true;
        Timers.StartSingleTimer(PublishTimerKey, new PublishCoalesced(), _publishCoalesce);
    }

ctor registration Receive<PublishCoalesced>(_ => { _publishPending = false; if (!_stopped) Publish(); }); and record internal sealed record PublishCoalesced;. In OnSeedCompleted (and OnSeedFailed's publish branch), before the immediate Publish(): Timers.Cancel(PublishTimerKey); _publishPending = false; (the seed snapshot already carries the buffered deltas).

• SiteAlarmLiveCacheService Props (line 228-236): append _options.LiveAlarmCachePublishCoalesce as the new last argument.

4. Run the filter, then the aggregator + service groups: dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter "SiteAlarmAggregatorActor|SiteAlarmLiveCache|CommunicationOptionsValidator" — fix any existing per-delta-publish assertions by constructing with TimeSpan.Zero (legacy behavior preserved for them).
5. Commit: git add src/ZB.MOM.WW.ScadaBridge.Communication tests/ZB.MOM.WW.ScadaBridge.Communication.Tests && git commit -m "perf(comm): coalesce live-alarm delta publishes (250ms window, 0 = legacy) (plan R2-02 T10)"

Task 11: Aggregator lifecycle — queued re-seed after reconnect + stream-generation stamp

Classification: standard Estimated implement time: ~5 min Parallelizable with: 1, 2, 5, 6, 7, 8, 9, 13, 14 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs (StartFanout lines 245-274; OnSeedCompleted/OnSeedFailed; OpenGrpcStream lines 407-433; GrpcAlarmStreamError record line 524 + handler lines 164-168)
• Test: tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs (append)

Depends on Task 10 (same files). N7.1: the post-failover StartFanout(isInitial: false) (line 472) no-ops if a reconcile fan-out is in flight — deltas missed between stream death and that snapshot's read-time stay stale up to 60 s despite the "never silently serve stale" comment. N7.2: a late error raced out of the previous cancelled stream is indistinguishable from a current-stream failure and burns retry budget / double-flips.

1. Failing tests:

    [Fact]
    public void ReseedRequestedDuringInFlightFanout_IsQueued_NotDropped()
    {
        var (actor, seed, sink, client) = CreateAggregator(publishCoalesce: TimeSpan.Zero);
        seed.CompleteNext();                       // initial seed done (CallCount 1)
        AwaitAssert(() => Assert.Equal(1, sink.Count));

        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 failover re-seed must not
        // be silently skipped. Pre-fix: StartFanout no-ops and CallCount stays 2 until
        // the next 60s reconcile tick.
        client.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 (actor, seed, sink, client) = CreateAggregator(publishCoalesce: TimeSpan.Zero);
        seed.CompleteNext();

        var firstSub = client.Subs.Single();
        firstSub.OnError(new Exception("real fault"));      // gen 1 error → flip, reconnect
        AwaitAssert(() => Assert.Equal(2, client.Subs.Count)); // gen 2 stream open

        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.
        Task.Delay(400).Wait();
        Assert.Equal(2, client.Subs.Count);
    }

2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter "IsQueued_NotDropped|PreviousStreamGeneration_IsIgnored" — expect FAIL.
3. Implement:
   • Queued re-seed: field private bool _reseedQueued;. StartFanout: if (_fanoutInFlight) { if (!isInitial) _reseedQueued = true; return; } (comment: a failover re-seed requested mid-fan-out must run right after — the in-flight snapshot's read-time predates the stream death, N7.1). In OnSeedCompleted and OnSeedFailed, after _fanoutInFlight = false; (and after the flush/publish): if (_reseedQueued) { _reseedQueued = false; StartFanout(isInitial: false); }.
   • Generation stamp: field private int _streamGeneration;. OpenGrpcStream: var generation = ++_streamGeneration; and the error callback becomes ex => self.Tell(new GrpcAlarmStreamError(ex, generation)); record becomes internal sealed record GrpcAlarmStreamError(Exception Exception, int Generation);; the handler (line 164-168) prepends: if (msg.Generation != _streamGeneration) { _log.Debug("Ignoring stale gRPC error from stream generation {0} (current {1})", msg.Generation, _streamGeneration); return; } (comment: the RpcException(Cancelled) filter at SiteStreamGrpcClient.cs:256-258 covers the normal path; a genuine socket fault can beat the cancel — N7.2).
4. Run the filter + --filter SiteAlarmAggregatorActor — expect PASS.
5. Commit: git add src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs && git commit -m "fix(comm): queue failover re-seed behind in-flight fan-out; stamp stream generation on gRPC errors (plan R2-02 T11)"

Task 12: Instance-injected ReconnectDelay/StabilityWindow (kill the process-global test seams)

Classification: small (mechanical; N7.3) Estimated implement time: ~4 min Parallelizable with: 1, 2, 5, 6, 7, 8, 9, 13, 14 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs (lines 61-69 statics → ctor-injected instance fields), src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs (Props.Create — pass production defaults explicitly)
• Test: tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs (ctor at lines 27-28 — construct with explicit values instead of mutating statics)

Depends on Task 11 (same files). The internal static mutable ReconnectDelay/StabilityWindow are process-global state parallel test classes can trample; every other tuning knob in this file is ctor-injected.

1. This is a refactor with existing coverage: the aggregator suite exercises both knobs (fast reconnect at 50 ms, long stability at 30 s). No new test — the "failing test" is the compile break that proves the statics are gone.
2. Implement: delete the static properties (lines 61-69); ctor gains TimeSpan reconnectDelay, TimeSpan stabilityWindow (explicit — expression trees reject optional args) stored in private readonly TimeSpan _reconnectDelay; private readonly TimeSpan _stabilityWindow;; replace the three usages (OpenGrpcStream line 419 stability timer, HandleGrpcError line 477 reconnect timer). SiteAlarmLiveCacheService.Props.Create passes TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(60) — annotated as the former static defaults. Test harness: remove the ctor static mutations (lines 27-28) and pass TimeSpan.FromMilliseconds(50), TimeSpan.FromSeconds(30) through its CreateAggregator helper.
3. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter "SiteAlarmAggregatorActor|SiteAlarmLiveCache" — expect PASS (identical timings, now instance-scoped).
4. Commit: git add src/ZB.MOM.WW.ScadaBridge.Communication tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs && git commit -m "chore(comm): ctor-inject aggregator reconnect/stability tuning, remove process-global statics (plan R2-02 T12)"

Task 13: Production caller for RemoveSiteAsync — dispose a deleted site's gRPC channels

Classification: small Estimated implement time: ~4 min Parallelizable with: 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs (HandleDeleteSite, lines 1574-1588)
• Test: tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementActorTests.cs (append)

N8 first half: SiteStreamGrpcClientFactory.RemoveSiteAsync (SiteStreamGrpcClientFactory.cs:95-102) is documented as "call it when a site record is deleted" and has no production caller — a deleted site's channels persist until process restart. ManagementActor.HandleDeleteSite is the single site-deletion path (the CLI and Central UI both dispatch DeleteSiteCommand through it).

1. Failing test (harness: ManagementActorTests DI ServiceCollection + Envelope(cmd, Roles.Administrator) pattern):

    [Fact]
    public async Task DeleteSite_DisposesTheSitesCachedGrpcChannels()
    {
        var siteRepo = Substitute.For<ISiteRepository>();
        siteRepo.GetSiteByIdAsync(7).Returns(new Site
            { Id = 7, SiteIdentifier = "site-7", Name = "Seven" });
        siteRepo.GetInstancesBySiteIdAsync(7).Returns(new List<Instance>());
        _services.AddScoped(_ => siteRepo);

        var factory = new TrackingGrpcFactory(); // subclass overriding CreateClient, as in SiteStreamGrpcClientFactoryTests
        var cached = (TrackingClient)factory.GetOrCreate("site-7", "http://node-a:8083");
        _services.AddSingleton<SiteStreamGrpcClientFactory>(factory);

        var actor = CreateActor();
        actor.Tell(Envelope(new DeleteSiteCommand(7), Roles.Administrator));
        ExpectMsg<ManagementResponse>(r => Assert.True(r.Success));

        await AwaitAssertAsync(() =>
        {
            Assert.True(cached.Disposed);
            Assert.Null(factory.TryGet("site-7", "http://node-a:8083"));
        });
    }

(Copy the TrackingGrpcFactory/TrackingClient doubles from tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryTests.cs — or reference them if visible.) 2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests --filter DeleteSite_DisposesTheSitesCachedGrpcChannels — expect FAIL (channels untouched). 3. Implement in HandleDeleteSite (after commService?.RefreshSiteAddresses();, line 1585):

        // Dispose the deleted site's cached gRPC channels — RemoveSiteAsync is the
        // factory's designed site-deletion disposal path and previously had no
        // production caller (arch review 02 round 2, N8): a deleted site's channels
        // (both node endpoints) otherwise persist until process restart. Null-safe:
        // test/composition roots without the factory skip it. Any live-alarm
        // aggregator for the site reconciles to an empty snapshot and is reaped by
        // the viewer linger (documented acceptance — see Component-Communication.md).
        var grpcFactory = sp.GetService<SiteStreamGrpcClientFactory>();
        if (grpcFactory is not null && site is not null)
            await grpcFactory.RemoveSiteAsync(site.SiteIdentifier);

(using ZB.MOM.WW.ScadaBridge.Communication.Grpc; — ManagementService already references Communication.) 4. Run the filter + dotnet test tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests --filter ManagementActorTests — expect PASS. 5. Commit: git add src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementActorTests.cs && git commit -m "fix(comm): site delete disposes the site's cached gRPC channels via RemoveSiteAsync (plan R2-02 T13)"

Task 14: Documented acceptance — standby-node aggregators + deleted-site viewer behavior

Classification: trivial (docs only) Estimated implement time: ~3 min Parallelizable with: 1, 2, 3, 5, 6, 7, 8, 9, 10, 11, 12, 13, 15 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs (doc lines 12-17), docs/requirements/Component-Communication.md (live-alarm-stream section)

Depends on Task 4 (shared Component-Communication.md — serialized). N8 second half: SetActorSystem is wired on every central node (AkkaHostedService.cs:429-430), so browsing the standby directly (port 9002, documented for diagnostics) starts a second, fully-functional aggregator + gRPC stream. The report calls it "harmless (read-only) but contradicting the [PERM] doc claim" — the fix is honest documentation, not gating (gating SetActorSystem behind the active check would break the diagnostic-browse path and buy nothing: the aggregator is read-only and linger-reaped).

1. ISiteAlarmLiveCache.cs doc (lines 12-17): replace "…lives only on the active central node…" with: "…purely in-memory, per-node, no EF entity/table/migration. In routine operation only the active node (behind Traefik) hosts viewers, so only it runs aggregators; browsing the standby node directly (diagnostic ports) starts an independent read-only aggregator there — accepted: bounded (one stream/site, viewer-capped), read-only, and torn down by the viewer linger. On a NodeA↔NodeB failover the new active node re-seeds from scratch."
2. Component-Communication.md, live-alarm-stream section — add an explicit "Accepted limitations (arch review 02 round 2, N8)" note covering (a) the standby-browse aggregator above, and (b) a site deleted while an Alarm Summary viewer is open: the viewer's aggregator reconciles to an empty snapshot until its viewers leave, then the linger stop reaps it; the site's gRPC channels are disposed at deletion (Task 13).
3. Verify: dotnet build ZB.MOM.WW.ScadaBridge.slnx (doc-comment change compiles) and grep — grep -rn "lives only on the active central node" src/ returns nothing.
4. Commit: git add src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs docs/requirements/Component-Communication.md && git commit -m "docs(comm): accept + document standby aggregators and deleted-site viewer behavior (plan R2-02 T14)"

Task 15: Single-endpoint sites can go live

Classification: small. MUTEX with PLAN-R2-07 (same SiteAlarmLiveCacheService.cs) — see the rule at the top. Estimated implement time: ~4 min Parallelizable with: 2, 4, 5, 6, 7, 8, 9, 13, 14 Files:

• Modify: src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs (ResolveSiteAsync, lines 294-302)
• Test: tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs (append)

Depends on Task 12 (file-serialized chain 10 → 12 → 15 on the live-alarm files). N9: ResolveSiteAsync requires both GrpcNodeAAddress and GrpcNodeBAddress; a single-node site silently never starts an aggregator and polls forever. The aggregator works fine flipping between one address — accept a single endpoint by reusing it for both slots (the NodeA↔NodeB flip degenerates to a reconnect against the same node).

1. Failing test (follow the file's CreateService(...) harness — it substitutes ISiteRepository and injects a CountingFactory):

    [Fact]
    public async Task SiteWithOnlyNodeAEndpoint_StartsAnAggregator_AgainstThatEndpoint()
    {
        var service = CreateService(TimeSpan.FromSeconds(30), out var factory,
            site: new Site
            {
                Id = 3, SiteIdentifier = "site-3",
                GrpcNodeAAddress = "http://only-node:8083",
                GrpcNodeBAddress = null, // single-endpoint site — pre-fix: never goes live
            });

        using var sub = service.Subscribe(3, onChanged: () => { });

        await AwaitAssertAsync(() =>
        {
            Assert.Equal(1, factory.CreatedCount); // pre-fix: 0 — ResolveSiteAsync bailed
            Assert.NotNull(factory.TryGet("site-3", "http://only-node:8083"));
        });
    }

(Adapt the harness call to CreateService's real signature; the essential arrangement is a site row with exactly one gRPC address.) 2. Run: dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter SiteWithOnlyNodeAEndpoint — expect FAIL (no client ever created; the service logged "has no gRPC node addresses"). 3. Implement — replace the both-required check (SiteAlarmLiveCacheService.cs:294-300):

        var nodeA = site.GrpcNodeAAddress;
        var nodeB = site.GrpcNodeBAddress;
        if (string.IsNullOrWhiteSpace(nodeA) && string.IsNullOrWhiteSpace(nodeB))
        {
            _logger.LogWarning(
                "Live alarm cache: site {SiteId} ({SiteIdentifier}) has no gRPC node addresses; " +
                "not starting an aggregator", siteId, site.SiteIdentifier);
            return null;
        }

        // Single-endpoint site (review 02 round 2, N9): reuse the sole configured
        // endpoint for both slots — the aggregator's NodeA↔NodeB failover flip
        // degenerates to a reconnect against the same node, which is exactly the
        // right behavior for a one-node site.
        var grpcA = !string.IsNullOrWhiteSpace(nodeA) ? nodeA! : nodeB!;
        var grpcB = !string.IsNullOrWhiteSpace(nodeB) ? nodeB! : nodeA!;
        return (site.SiteIdentifier, grpcA, grpcB);

4. Run the filter + dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter SiteAlarmLiveCache — expect PASS.
5. Commit: git add src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs && git commit -m "fix(comm): single-gRPC-endpoint sites can go live (flip degenerates to same-node reconnect) (plan R2-02 T15)"

---

Dependencies on other plans

• PLAN-R2-07 (UI/Management/Security round 2) — owns every AlarmSummary.razor page-side finding and the sticky-IsLive change in SiteAlarmLiveCacheService.cs. Binding MUTEX (see top of plan): Tasks 10, 12, 15 must not run concurrently with R2-07's live-alarm slice. Everything else here is file-disjoint from R2-07.
• PLAN-01 heritage — Task 1 relocates (by delegation, API-preserving) the oldest-Up logic PLAN-01 shipped in Host/Health/ClusterActivityEvaluator.cs; the PLAN-01 TwoNodeClusterFixture is reused (additively extended) by Task 2. No live round-2 plan owns those files.
• Contract locks — the new SfBufferSnapshotChunk/SfBufferResyncAck messages ride intra-site Akka remoting, NOT ClusterClient, so ClusterClientContractLockTests (PLAN-08 T9) needs no update; the equivalent wire pins land in ResyncWireSerializationPinTests (Task 5). If a later change routes any resync message over ClusterClient, it MUST be added to the frozen contract-lock matrix.

Execution order

P0 (the Critical, strictly ordered): 1 → 2 (red repro) → 3 (repro goes green) → 4. Do not ship a build between 2 and 3.

Critical path (file-serialized chains):

• SiteReplicationActor.cs: 3 → 5 → 6 → 7
• AkkaHostedService.cs: 3 → 4
• Live-alarm files (SiteAlarmAggregatorActor.cs / SiteAlarmLiveCacheService.cs): 10 → 11 → 12 → 15 (MUTEX with PLAN-R2-07 throughout)
• Component-Communication.md: 4 → 14
• StoreAndForwardService.cs: 8 standalone; validator: 9 standalone; ManagementActor.cs: 13 standalone

P1 (the High): 5 → 6 → 7 immediately after P0. P2: 8, 9, 10, 11, 12, 13, 14, 15 in any dependency-respecting order; 8/9/13/14 are safe to run in parallel with the P0/P1 chains (file-disjoint).

After each task: build the touched projects + the named per-project --filter command only. At plan end: dotnet build ZB.MOM.WW.ScadaBridge.slnx, then dotnet test ZB.MOM.WW.ScadaBridge.slnx (the slnx now includes CLI.Tests), and rebuild the docker cluster (bash docker/deploy.sh) since Host/actor runtime changed.