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

# Communication & Store-and-Forward Fix Implementation Plan

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

**Goal:** Fix every finding in `archreview/02-communication-store-and-forward.md` — the ClusterClient recreate crash-loop, the ungated standby S&F delivery pipeline, the shared-gRPC-channel disposal, the collapsing retry sweep, replication ordering/upsert/observability, notification parking/blocking semantics, corrupt-payload handling, and the SQLite/serializer/observability hygiene items.

**Architecture:** The fixes stay inside the existing seams: `CentralCommunicationActor`/`DefaultSiteClientFactory` get collision-free generation-suffixed actor names; `StoreAndForwardService` gets an injectable active-node delivery gate (a `Func<bool>` seam that plan 01's shared singleton-host helper will later plug into), a bounded/short-circuiting/parallel-lane sweep, and a defer-to-sweep enqueue path so `Notify.Send` never blocks a script thread; `SiteStreamGrpcClientFactory` is re-keyed by `(site, endpoint)` so per-session failover cannot dispose shared channels; `ReplicationService` dispatches inline (ordered) with Warning-level observability and upsert-based standby applies, plus a peer-join buffer resync as anti-entropy.

**Tech Stack:** C#/.NET 10, Akka.NET (TestKit.Xunit2, ClusterClient, DistributedPubSub), Microsoft.Data.Sqlite, Grpc.Net, xUnit + NSubstitute. Build: `dotnet build ZB.MOM.WW.ScadaBridge.slnx`. Test per-project: `dotnet test tests/<project>`.

## Findings Coverage

| Report finding | Severity | Task(s) |
|---|---|---|
| ClusterClient recreate-on-address-change → name collision → permanent restart loop | Critical | 1, 2 |
| `siteId` interpolated unvalidated into actor name (secondary, same lines) | Critical (sub) | 1 |
| Standby site node runs full S&F delivery pipeline (systematic duplicate delivery) | High | 3, 4 |
| Debug-session gRPC failover disposes shared channel under other sessions; `CleanupGrpc` creates channels during teardown | High | 5, 6 |
| Retry sweep serial, unbounded, single-laned — collapses under backlog | High (perf) | 7, 8, 9 |
| Replication operations applied out of order (`Task.Run` fire-and-forget) | Medium | 10 |
| Replication failures logged at Debug — dead channel invisible | Medium | 10 |
| Park/Requeue replication is a blind UPDATE — lost on standby missing the row | Medium | 11 |
| Notifications park after ~25 min of central unreachability; `maxRetries: 0` escape hatch unused | Medium | 13 |
| Corrupt buffered notification discarded and reported as delivered | Medium | 14 |
| Spec-vs-code drift: standby-passive model; `Notify.Send` blocks script up to 30 s | Medium | 4 (doc), 12, 13 |
| No explicit Akka serializer — wire format unpinned | Medium (perf) | 15 (round-trip pin tests). Serializer *swap* Deferred: changing the wire format breaks rolling cross-version central↔site compat; needs its own migration plan (flagged for a follow-up design session) |
| Central heartbeat state not replicated to peer central node | Low | 16 |
| `SubscribeInstance` concurrency-limit check-then-act (comment only) | Low | 22 |
| S&F SQLite: no WAL, no busy_timeout | Low | 17 |
| Due-check does per-row `julianday()` string parsing | Low | 18 |
| Audit-observer notification awaited inline in the sweep | Low | 19 |
| `StreamRelayActor.WriteToChannel` dead branch + silent DropOldest loss | Low | 22 |
| Duplicated Ask-timeout constants between the two ingest transports | Low | 22 |
| Heartbeat `IsActive` leader-check vs oldest-node singleton placement | Low | **Deferred → plan 01** (owns the shared active-node/singleton-host helper; Task 3's `Func<bool>` seam is designed to accept it) |
| `TransportHeartbeatInterval` double duty (transport FD + app heartbeat) | Low | 22 |
| U1: Dual transports for the same ingest operations | Underdev | **Won't-fix here** — consolidation spans the central ingest actors (another reviewer's domain); Task 22's shared timeout constant removes the copy-paste drift; consolidation flagged as a scheduled follow-up in the Task 22 doc note |
| U2: No real-collaborator test for ClusterClient cache lifecycle | Underdev | 2 |
| U3: No test/design treatment of standby-node sweep behavior | Underdev | 3, 4 (unit-level); real two-node kill rig → plan 01 dependency |
| U4: Replication has no anti-entropy/resync | Underdev | 20, 21 |
| U5: Parked rows have unbounded retention, no aging signal | Underdev | 23 |
| U6: Reconciliation cursor edge (`>` vs `>=`) unpinned | Underdev | 24 |
| U7: `DebugStreamService._sessions` lost silently on central restart | Underdev | 22 (documented as accepted limitation in `Component-Communication.md`) |

---

### Task 1: Generation-suffixed, sanitized ClusterClient actor names

**Classification:** high-risk (actor lifecycle/naming)
**Estimated implement time:** ~5 min
**Parallelizable with:** 3, 5, 10, 14, 15, 24
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs` (lines 33-41, `DefaultSiteClientFactory`)
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DefaultSiteClientFactoryTests.cs` (create)

**Design decision — generation suffix over await-termination:** `Context.Stop` is async and the name stays reserved until the old ClusterClient fully terminates. Watching `Terminated` and deferring recreation requires a pending-recreate state machine in the actor and *lengthens* the routing gap (envelopes for that site are dropped until the new client exists either way). A per-incarnation generation suffix makes creation collision-free by construction, is synchronous, and mirrors the existing repo precedent (`SiteStreamGrpcServer._actorCounter`, `SiteStreamGrpcServer.cs:36,259`). Sanitization handles validity; the counter guarantees uniqueness even if two site ids sanitize identically.

1. Write the failing test:

```csharp
// tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DefaultSiteClientFactoryTests.cs
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Configuration;
using Akka.TestKit.Xunit2;
using Xunit;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;

public class DefaultSiteClientFactoryTests : TestKit
{
    private static readonly Config TestConfig = ConfigurationFactory.ParseString(@"
        akka.actor.provider = cluster
        akka.remote.dot-netty.tcp.port = 0
        akka.remote.dot-netty.tcp.hostname = localhost")
        .WithFallback(ClusterClientReceptionist.DefaultConfig());

    public DefaultSiteClientFactoryTests() : base(TestConfig) { }

    private static ImmutableHashSet<ActorPath> Contacts() =>
        ImmutableHashSet.Create(ActorPath.Parse("akka.tcp://other@localhost:2552/system/receptionist"));

    [Fact]
    public void Create_TwiceForSameSite_DoesNotCollide()
    {
        var factory = new DefaultSiteClientFactory();
        var first = factory.Create(Sys, "site-a", Contacts());
        var second = factory.Create(Sys, "site-a", Contacts()); // pre-fix: InvalidActorNameException
        Assert.NotEqual(first.Path, second.Path);
    }

    [Theory]
    [InlineData("site/with/slashes")]
    [InlineData("site with spaces")]
    [InlineData("näme#!")]
    public void Create_WithUnsafeSiteId_SanitizesAndSucceeds(string siteId)
    {
        var factory = new DefaultSiteClientFactory();
        var client = factory.Create(Sys, siteId, Contacts()); // pre-fix: InvalidActorNameException
        Assert.NotNull(client);
    }

    [Theory]
    [InlineData("plant-01", "plant-01")]
    [InlineData("a/b c", "a_b_c")]
    [InlineData("", "site")]
    public void SanitizeForActorName_ProducesValidPathElement(string input, string expected)
    {
        Assert.Equal(expected, DefaultSiteClientFactory.SanitizeForActorName(input));
        Assert.True(ActorPath.IsValidPathElement(
            DefaultSiteClientFactory.SanitizeForActorName(input)));
    }
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter DefaultSiteClientFactoryTests` — expect FAIL (collision throws / sanitize method missing).
3. Implement in `CentralCommunicationActor.cs`, replacing the `DefaultSiteClientFactory` body:

```csharp
public class DefaultSiteClientFactory : ISiteClientFactory
{
    /// <summary>
    /// Per-incarnation generation counter. Context.Stop of the previous client is
    /// asynchronous — its actor name stays reserved until termination completes —
    /// so a same-named recreate in the same message handling throws
    /// InvalidActorNameException. A generation suffix makes every incarnation's
    /// name unique by construction (mirrors SiteStreamGrpcServer._actorCounter).
    /// </summary>
    private long _generation;

    /// <inheritdoc />
    public IActorRef Create(ActorSystem system, string siteId, ImmutableHashSet<ActorPath> contacts)
    {
        var settings = ClusterClientSettings.Create(system).WithInitialContacts(contacts);
        var name = $"site-client-{SanitizeForActorName(siteId)}-{Interlocked.Increment(ref _generation)}";
        return system.ActorOf(ClusterClient.Props(settings), name);
    }

    /// <summary>
    /// Maps an arbitrary SiteIdentifier onto a valid Akka actor-path element:
    /// letters/digits/'-'/'_' pass through, everything else becomes '_'. Uniqueness
    /// is NOT required here (the generation suffix guarantees it); only validity is.
    /// </summary>
    internal static string SanitizeForActorName(string siteId)
    {
        if (string.IsNullOrEmpty(siteId)) return "site";
        var sanitized = new string(siteId
            .Select(c => char.IsAsciiLetterOrDigit(c) || c is '-' or '_' ? c : '_')
            .ToArray());
        return ActorPath.IsValidPathElement(sanitized) ? sanitized : "site";
    }
}
```

   (`internal` is visible to the test project via the existing `InternalsVisibleTo` in the csproj.)
4. Run the filter again — expect PASS. Also run the full project: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests`.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DefaultSiteClientFactoryTests.cs && git commit -m "fix(communication): generation-suffixed, sanitized ClusterClient actor names to prevent recreate name collision"`

### Task 2: Guard client creation in the refresh loop + real-factory address-edit lifecycle test

**Classification:** high-risk (actor restart semantics)
**Estimated implement time:** ~5 min
**Parallelizable with:** 3, 5, 7, 10, 14
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs` (`HandleSiteAddressCacheLoaded`, lines 507-526)
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs` (create)

Depends on Task 1 (same file, uses the fixed factory).

1. Write the failing test — the address-edit scenario with the **real** `DefaultSiteClientFactory` (this is the exact production path the report says no test exercises; follow the DI-stub pattern from `CentralCommunicationActorTests` and the TestKit config from Task 1):

```csharp
// tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs
public class CentralCommunicationActorClientLifecycleTests : TestKit
{
    // same TestConfig as DefaultSiteClientFactoryTests

    private static SiteAddressCacheLoaded Load(string siteId, params string[] addrs) =>
        new(new Dictionary<string, IReadOnlyList<string>>
            { [siteId] = addrs.ToList().AsReadOnly() });

    [Fact]
    public void AddressEdit_RecreatesClient_WithoutRestartLoop()
    {
        var sp = new ServiceCollection().BuildServiceProvider(); // no repo needed: we inject cache loads directly
        var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(
            sp, new DefaultSiteClientFactory(), null)));

        // First load creates the client; second load (edited NodeA address) stops
        // the old one and creates a replacement in the same message handling.
        // Pre-fix this throws InvalidActorNameException and restarts the actor.
        EventFilter.Exception<InvalidActorNameException>().Expect(0, () =>
        {
            actor.Tell(Load("site-a", "akka.tcp://scadabridge@node-a:8081"));
            actor.Tell(Load("site-a", "akka.tcp://scadabridge@node-a-edited:8081"));
            actor.Tell(Load("site-a", "akka.tcp://scadabridge@node-a:8081")); // and back — third generation
        });

        // The actor must still be alive and routing (not crash-looping):
        // an envelope for an unknown site produces the "No ClusterClient" warning,
        // proving the Receive pipeline is healthy.
        EventFilter.Warning(contains: "No ClusterClient for site").ExpectOne(() =>
            actor.Tell(new SiteEnvelope("unknown-site", new object())));
    }

    [Fact]
    public void FactoryThrow_SkipsSite_DoesNotCrashActor()
    {
        var throwingFactory = Substitute.For<ISiteClientFactory>();
        throwingFactory.Create(Arg.Any<ActorSystem>(), Arg.Any<string>(), Arg.Any<ImmutableHashSet<ActorPath>>())
            .Returns(_ => throw new InvalidOperationException("boom"));
        var sp = new ServiceCollection().BuildServiceProvider();
        var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(
            sp, throwingFactory, null)));

        EventFilter.Error(contains: "Failed to create ClusterClient").ExpectOne(() =>
            actor.Tell(Load("site-a", "akka.tcp://scadabridge@node-a:8081")));

        // Actor survived — a subsequent load with a healthy factory-free path still processes.
        EventFilter.Warning(contains: "No ClusterClient for site").ExpectOne(() =>
            actor.Tell(new SiteEnvelope("site-a", new object())));
    }
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter CentralCommunicationActorClientLifecycleTests` — expect FAIL (`FactoryThrow` case escapes the handler and restarts the actor; pre-Task-1 the first case also fails).
3. Implement — in `HandleSiteAddressCacheLoaded`, wrap the create (currently line 520) so a failure logs, cleans the entry, and skips the site instead of crashing the actor:

```csharp
            // Stop old client if addresses changed
            if (_siteClients.ContainsKey(siteId))
            {
                _log.Info("Updating ClusterClient for site {0} (addresses changed)", siteId);
                Context.Stop(_siteClients[siteId].Client);
                // Remove now: if the replacement create below fails, a stale entry
                // would route envelopes to a stopping actor.
                _siteClients.Remove(siteId);
            }

            IActorRef client;
            try
            {
                client = _siteClientFactory.Create(Context.System, siteId, contactPaths);
            }
            catch (Exception ex)
            {
                _log.Error(ex,
                    "Failed to create ClusterClient for site {0}; site is unroutable until the next refresh",
                    siteId);
                continue;
            }
            _siteClients[siteId] = (client, contactStrings);
```

4. Run the filter — expect PASS. Run full project tests.
5. Commit: `git add -A src/ZB.MOM.WW.ScadaBridge.Communication tests/ZB.MOM.WW.ScadaBridge.Communication.Tests && git commit -m "fix(communication): guard per-site ClusterClient creation; add real-factory address-edit lifecycle test"`

### Task 3: Active-node delivery gate on the S&F retry sweep

**Classification:** high-risk (failover semantics)
**Estimated implement time:** ~5 min
**Parallelizable with:** 1, 2, 5, 14, 15, 16, 24
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs` (fields near line 85; `RetryPendingMessagesAsync`, lines 563-589)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs` (append)

The gate is a `Func<bool>` seam re-evaluated on **every sweep tick** — failover resumes delivery within one `RetryTimerInterval` (10 s) with no re-wiring. Plan 01 owns the shared active-node/singleton-host helper; this seam is where it plugs in (Task 4 wires the interim leader-check).

1. Write the failing tests (append to `StoreAndForwardServiceTests.cs`, reusing its temp-SQLite construction pattern):

```csharp
[Fact]
public async Task RetrySweep_SkipsDelivery_WhenDeliveryGateReportsStandby()
{
    var service = CreateService(); // existing helper pattern: storage + options + service
    var delivered = 0;
    service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
        _ => { delivered++; return Task.FromResult(true); });
    service.SetDeliveryGate(() => false); // standby

    await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "target-x", "{}",
        attemptImmediateDelivery: false);
    await service.RetryPendingMessagesAsync();

    Assert.Equal(0, delivered);
    Assert.NotNull(await service.GetMessageByIdAsync(
        (await GetOnlyPendingId(service)))); // row untouched, still Pending
}

[Fact]
public async Task RetrySweep_ResumesDelivery_WhenGateFlipsActive()
{
    var service = CreateService();
    var delivered = 0;
    var active = false;
    service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
        _ => { delivered++; return Task.FromResult(true); });
    service.SetDeliveryGate(() => active);

    await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "target-x", "{}",
        attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);

    await service.RetryPendingMessagesAsync();
    Assert.Equal(0, delivered);

    active = true; // failover: this node became active
    await service.RetryPendingMessagesAsync();
    Assert.Equal(1, delivered);
}

[Fact]
public async Task RetrySweep_TreatsThrowingGateAsStandby()
{
    var service = CreateService();
    var delivered = 0;
    service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
        _ => { delivered++; return Task.FromResult(true); });
    service.SetDeliveryGate(() => throw new InvalidOperationException("cluster not ready"));

    await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}",
        attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);
    await service.RetryPendingMessagesAsync(); // must not throw
    Assert.Equal(0, delivered);
}
```

   Note: `attemptImmediateDelivery: false` sets `LastAttemptAt = now`, so pass `retryInterval: TimeSpan.Zero` (or use the storage `retry_interval_ms = 0` always-due path) to make the row due immediately.
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter RetrySweep_` — expect FAIL (`SetDeliveryGate` missing).
3. Implement in `StoreAndForwardService`:

```csharp
    /// <summary>
    /// Active-node delivery gate. When set, the retry sweep runs ONLY when the
    /// gate reports true — the standby node applies replicated buffer operations
    /// but must never deliver (Component-StoreAndForward.md "standby-passive"
    /// model). Re-evaluated on every sweep tick so a failover resumes delivery
    /// within one RetryTimerInterval without re-wiring. Null (tests, central
    /// hosts) preserves the ungated legacy behaviour. A throwing gate is treated
    /// as standby — safe-by-default, mirroring SiteCommunicationActor's
    /// DefaultIsActiveCheck fallback.
    /// </summary>
    private Func<bool>? _deliveryGate;

    /// <summary>Installs the active-node delivery gate (see <see cref="_deliveryGate"/>).</summary>
    public void SetDeliveryGate(Func<bool> gate) => _deliveryGate = gate;
```

   And at the top of `RetryPendingMessagesAsync`, immediately after the `_retryInProgress` CAS succeeds (inside the `try`):

```csharp
            var gate = _deliveryGate;
            if (gate != null)
            {
                bool isActive;
                try { isActive = gate(); }
                catch (Exception ex)
                {
                    _logger.LogWarning(ex,
                        "S&F delivery gate threw; treating this node as standby for this sweep");
                    isActive = false;
                }
                if (!isActive)
                {
                    _logger.LogDebug("S&F retry sweep skipped: this node is not the active site node");
                    return;
                }
            }
```

4. Run the filter — expect PASS. Run `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests`.
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(store-and-forward): gate the retry sweep behind an active-node delivery gate (standby must be passive)"`

### Task 4: Wire the delivery gate in the Host + align the S&F design doc

**Classification:** small (wiring + docs; behavior covered by Task 3's tests)
**Estimated implement time:** ~3 min
**Parallelizable with:** 5, 6, 10, 11, 14
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs` (in `RegisterSiteActorsAsync`, immediately after `await storeAndForwardService.StartAsync();` at line 981)
- Modify: `docs/requirements/Component-StoreAndForward.md` (Persistence section, the failover paragraphs at lines ~77-80)

Depends on Task 3. **Dependency on plan 01:** plan 01 delivers the shared active-node/singleton-host helper (and the leader-vs-oldest correction). Until it lands, wire the same leader+Up check the repo already uses (`ActiveNodeGate`, `SiteCommunicationActor.DefaultIsActiveCheck`); the lambda body is the single swap point.

1. Implement the wiring in `AkkaHostedService.RegisterSiteActorsAsync`:

```csharp
            await storeAndForwardService.StartAsync();

            // Standby must be passive: only the active site node runs the S&F
            // delivery sweep. Without this gate BOTH nodes deliver the same
            // replicated Pending rows — systematic duplicate external calls and
            // DB writes (arch review 02, Stability #2). Re-evaluated per sweep
            // tick so failover resumes delivery within one RetryTimerInterval.
            // NOTE: leader+Up mirrors ActiveNodeGate / ActiveNodeHealthCheck /
            // SiteCommunicationActor.DefaultIsActiveCheck. When the shared
            // singleton-host helper from the cluster-infrastructure fix plan
            // lands, replace this lambda body with it.
            var gateSystem = _actorSystem!;
            storeAndForwardService.SetDeliveryGate(() =>
            {
                var cluster = Akka.Cluster.Cluster.Get(gateSystem);
                var self = cluster.SelfMember;
                return self.Status == Akka.Cluster.MemberStatus.Up
                    && cluster.State.Leader != null
                    && cluster.State.Leader == self.Address;
            });
```

2. Update `docs/requirements/Component-StoreAndForward.md` Persistence/failover text: replace the sentence "On failover, the new active node resumes delivery from its local copy." with an explicit statement that (a) the delivery sweep is **gated to the active node** and the standby only applies replicated operations, and (b) duplicate deliveries are confined to the failover window (an in-flight delivery whose `Remove` was not yet replicated), not steady-state operation.
3. Build: `dotnet build ZB.MOM.WW.ScadaBridge.slnx` — expect success. Run `dotnet test tests/ZB.MOM.WW.ScadaBridge.Host.Tests` (regression only).
4. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs docs/requirements/Component-StoreAndForward.md && git commit -m "fix(host): wire S&F active-node delivery gate on site nodes; align S&F doc standby-passive wording"`

### Task 5: Key the gRPC client cache by (site, endpoint)

**Classification:** standard
**Estimated implement time:** ~5 min
**Parallelizable with:** 1, 2, 3, 10, 14, 15
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs` (whole cache surface, lines 14, 52-137)
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryTests.cs` (modify — the dispose-on-endpoint-change assertions invert), `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryDisposeTests.cs` (modify)

Both node channels coexist per site; per-session failover never disposes shared state. Site *removal* (`RemoveSiteAsync`) disposes all of a site's entries — that remains the only shared-disposal path. Trade-off (documented in the class doc): an edited gRPC address leaves the old endpoint's idle channel cached until site removal or process shutdown — bounded at a handful of entries per site (2 nodes + rare edits), each an idle HTTP/2 channel.

1. Modify the existing factory tests (they use the `protected virtual CreateClient` tracking-client override): the test that asserts "requesting a different endpoint disposes the cached client" becomes:

```csharp
[Fact]
public void GetOrCreate_DifferentEndpointSameSite_KeepsBothClientsAlive()
{
    var factory = new TrackingFactory(LoggerFactory);
    var a = factory.GetOrCreate("site-1", "http://node-a:8083");
    var b = factory.GetOrCreate("site-1", "http://node-b:8083");

    Assert.NotSame(a, b);
    Assert.False(((TrackingClient)a).Disposed); // pre-fix: a is disposed here
    Assert.Same(a, factory.GetOrCreate("site-1", "http://node-a:8083")); // still cached
}

[Fact]
public void TryGet_ReturnsCachedClientOrNull_WithoutCreating()
{
    var factory = new TrackingFactory(LoggerFactory);
    Assert.Null(factory.TryGet("site-1", "http://node-a:8083"));
    var a = factory.GetOrCreate("site-1", "http://node-a:8083");
    Assert.Same(a, factory.TryGet("site-1", "http://node-a:8083"));
    Assert.Null(factory.TryGet("site-1", "http://node-b:8083"));
    Assert.Equal(1, factory.CreatedCount); // TryGet never creates
}

[Fact]
public async Task RemoveSiteAsync_DisposesAllEndpointsForTheSite_OnlyThatSite()
{
    var factory = new TrackingFactory(LoggerFactory);
    var a = (TrackingClient)factory.GetOrCreate("site-1", "http://node-a:8083");
    var b = (TrackingClient)factory.GetOrCreate("site-1", "http://node-b:8083");
    var other = (TrackingClient)factory.GetOrCreate("site-2", "http://node-a:8083");

    await factory.RemoveSiteAsync("site-1");
    Assert.True(a.Disposed);
    Assert.True(b.Disposed);
    Assert.False(other.Disposed);
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter SiteStreamGrpcClientFactory` — expect FAIL.
3. Implement:

```csharp
    private readonly ConcurrentDictionary<(string Site, string Endpoint), SiteStreamGrpcClient> _clients = new();

    public virtual SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint) =>
        _clients.GetOrAdd((siteIdentifier, grpcEndpoint), _ => CreateClient(grpcEndpoint));

    /// <summary>
    /// Returns the cached client for (site, endpoint), or null — never creates.
    /// Teardown paths (DebugStreamBridgeActor.CleanupGrpc / failed-stream
    /// unsubscribe) use this so cleanup can never open a fresh channel.
    /// </summary>
    public virtual SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint) =>
        _clients.TryGetValue((siteIdentifier, grpcEndpoint), out var client) ? client : null;

    public async Task RemoveSiteAsync(string siteIdentifier)
    {
        foreach (var key in _clients.Keys.Where(k => k.Site == siteIdentifier).ToList())
        {
            if (_clients.TryRemove(key, out var client))
                await client.DisposeAsync();
        }
    }
```

   `DisposeAsync`/`Dispose` iterate `_clients.Values` unchanged. Update the class `<summary>` (lines 40-48) — GetOrCreate no longer disposes on endpoint mismatch; both node channels coexist; site removal is the disposal path; note the idle-stale-channel trade-off. Fix any `RemoveSiteAsync` caller expectations (grep: only tests today).
4. Run the filter — expect PASS. Run the whole project (DebugStreamBridgeActorTests may still pass pre-Task-6 since GetOrCreate keeps working).
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc && git commit -m "fix(communication): key gRPC client cache by (site, endpoint) so per-session failover cannot dispose shared channels"`

### Task 6: Endpoint-safe unsubscribe in DebugStreamBridgeActor + concurrent-session regression test

**Classification:** high-risk (actor/stream lifecycle)
**Estimated implement time:** ~5 min
**Parallelizable with:** 3, 4, 7, 10, 14
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs` (`HandleGrpcError` lines 463-470, `CleanupGrpc` lines 486-495)
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs` (append)

Depends on Task 5 (`TryGet`).

1. Write the failing test (use the existing tracking-factory pattern in `DebugStreamBridgeActorTests`):

```csharp
[Fact]
public void SessionFailover_DoesNotDisposeOrTouch_OtherEndpointClient()
{
    var factory = new TrackingFactory(LoggerFactory);
    // Session A on node-a; session B's healthy client on node-b already cached.
    var bClient = (TrackingClient)factory.GetOrCreate("site-1", "http://node-b:8083");

    var bridge = CreateBridge(factory, nodeA: "http://node-a:8083", nodeB: "http://node-b:8083");
    bridge.Tell(new GrpcStreamError(new Exception("stream fault"))); // A flips a->b

    AwaitAssert(() =>
    {
        Assert.False(bClient.Disposed);          // pre-Task-5 fix: disposed
        Assert.Equal(0, bClient.UnsubscribeCallsForOtherCorrelations); // untouched
    });
}

[Fact]
public void CleanupGrpc_NeverCreatesAChannel()
{
    var factory = new TrackingFactory(LoggerFactory);
    var bridge = CreateBridge(factory, nodeA: "http://node-a:8083", nodeB: "http://node-b:8083");
    var createdBefore = factory.CreatedCount; // 1: PreStart opened node-a stream

    bridge.Tell(new StopDebugStream());       // teardown path
    AwaitAssert(() => Assert.Equal(createdBefore, factory.CreatedCount)); // pre-fix: GetOrCreate in CleanupGrpc creates
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter "SessionFailover_DoesNotDispose|CleanupGrpc_NeverCreates"` — expect FAIL on the create-count assertion.
3. Implement — replace both `GetOrCreate` teardown uses with `TryGet`:

```csharp
        // HandleGrpcError (was lines 468-470):
        var previousEndpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
        // TryGet, not GetOrCreate: unsubscribing a failed stream must never open a
        // fresh channel (and, pre-(site,endpoint)-keying, must never dispose another
        // session's healthy channel). Absent client => the channel is already gone
        // and the site-side relay will be reaped by keepalive/session-lifetime.
        _grpcFactory.TryGet(_siteIdentifier, previousEndpoint)?.Unsubscribe(_correlationId);
```

```csharp
    private void CleanupGrpc()
    {
        _grpcCts?.Cancel();
        _grpcCts?.Dispose();
        _grpcCts = null;

        var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
        _grpcFactory.TryGet(_siteIdentifier, endpoint)?.Unsubscribe(_correlationId);
    }
```

4. Run the filter, then the full project — expect PASS.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs && git commit -m "fix(communication): debug-stream teardown/failover unsubscribes via TryGet, never creates or disposes shared channels"`

### Task 7: Bound the retry sweep with a batch LIMIT

**Classification:** standard
**Estimated implement time:** ~4 min
**Parallelizable with:** 1, 2, 5, 6, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs` (`GetMessagesForRetryAsync`, lines 183-203), `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptions.cs`, `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs` (line 571)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs` (append), `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsTests.cs` (append)

1. Failing tests:

```csharp
// StoreAndForwardStorageTests
[Fact]
public async Task GetMessagesForRetry_HonorsLimit_OldestFirst()
{
    for (var i = 0; i < 5; i++)
        await _storage.EnqueueAsync(NewPendingMessage($"m{i}", createdAt: Base.AddSeconds(i)));

    var page = await _storage.GetMessagesForRetryAsync(limit: 3);
    Assert.Equal(3, page.Count);
    Assert.Equal(new[] { "m0", "m1", "m2" }, page.Select(m => m.Id));
}

// StoreAndForwardOptionsTests
[Fact]
public void SweepBatchLimit_Defaults_To_500() =>
    Assert.Equal(500, new StoreAndForwardOptions().SweepBatchLimit);
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "GetMessagesForRetry_HonorsLimit|SweepBatchLimit_Defaults"` — expect FAIL (no `limit` parameter / no option).
3. Implement:
   - `StoreAndForwardOptions`: `/// <summary>Maximum due rows loaded per retry sweep. Bounds sweep memory and duration under backlog; the remainder is picked up next tick (10 s). 0 = unlimited (legacy).</summary> public int SweepBatchLimit { get; set; } = 500;`
   - `GetMessagesForRetryAsync(int limit = 0)`: append `LIMIT @limit` when `limit > 0` (`cmd.CommandText += "\n            LIMIT @limit"; cmd.Parameters.AddWithValue("@limit", limit);`). Keep `ORDER BY created_at ASC` (oldest-first fairness).
   - `RetryPendingMessagesAsync` line 571: `var messages = await _storage.GetMessagesForRetryAsync(_options.SweepBatchLimit);`
4. Run the filter, then the full project — expect PASS.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests && git commit -m "perf(store-and-forward): bound retry sweep with SweepBatchLimit (default 500) on the due-rows query"`

### Task 8: Per-target short-circuit in the retry sweep

**Classification:** high-risk (retry-count semantics)
**Estimated implement time:** ~5 min
**Parallelizable with:** 2, 5, 6, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs` (`RetryPendingMessagesAsync` lines 563-589, `RetryMessageAsync` signature/returns, lines 591-720)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs` (append)

Depends on Task 7 (same methods). After the first *transient* failure to a `(category, target)` in a sweep, remaining messages for that pair are skipped — N messages to a dead target cost one timeout per sweep, not N. Skipped messages are NOT counted as retry attempts (their `RetryCount`/`LastAttemptAt` are untouched), so the park horizon reflects real attempts.

1. Failing test:

```csharp
[Fact]
public async Task RetrySweep_ShortCircuitsTarget_AfterFirstTransientFailure()
{
    var service = CreateService();
    var attemptsByTarget = new Dictionary<string, int>();
    service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, msg =>
    {
        attemptsByTarget[msg.Target] = attemptsByTarget.GetValueOrDefault(msg.Target) + 1;
        if (msg.Target == "dead-target") throw new TimeoutException("down");
        return Task.FromResult(true);
    });

    for (var i = 0; i < 3; i++)
        await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "dead-target", "{}",
            attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);
    await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "healthy-target", "{}",
        attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);

    await service.RetryPendingMessagesAsync();

    Assert.Equal(1, attemptsByTarget["dead-target"]);    // pre-fix: 3
    Assert.Equal(1, attemptsByTarget["healthy-target"]); // healthy lane unaffected
}

[Fact]
public async Task RetrySweep_SkippedMessages_DoNotAccrueRetryCount()
{
    var service = CreateService();
    service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
        _ => throw new TimeoutException("down"));
    await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "dead", "{}", maxRetries: 5,
        attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);
    var skippedId = (await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "dead", "{}",
        maxRetries: 5, attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero)).MessageId;

    await service.RetryPendingMessagesAsync();
    var skipped = await service.GetMessageByIdAsync(skippedId);
    Assert.Equal(0, skipped!.RetryCount); // only the attempted message accrued a retry
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "ShortCircuitsTarget|DoNotAccrueRetryCount"` — expect FAIL.
3. Implement:
   - Add `internal enum RetryOutcome { Delivered, Parked, TransientFailure, Skipped }` and change `RetryMessageAsync` to `private async Task<RetryOutcome> RetryMessageAsync(StoreAndForwardMessage message)` returning `Delivered` on success, `Parked` on both park branches, `TransientFailure` on the non-park transient branch, `Skipped` on no-handler/CAS-skip paths.
   - In `RetryPendingMessagesAsync`:

```csharp
            var failedTargets = new HashSet<(StoreAndForwardCategory, string)>();
            foreach (var message in messages)
            {
                // One transient failure per (category, target) per sweep: the target
                // is down — burning a full timeout per remaining message serializes
                // the sweep into hours under backlog (arch review 02, Performance #1).
                // Skipped rows keep their RetryCount/LastAttemptAt untouched.
                if (failedTargets.Contains((message.Category, message.Target)))
                    continue;

                var outcome = await RetryMessageAsync(message);
                if (outcome == RetryOutcome.TransientFailure)
                    failedTargets.Add((message.Category, message.Target));
            }
```

4. Run the filter + full project — expect PASS.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs && git commit -m "perf(store-and-forward): short-circuit a (category,target) lane after its first transient failure per sweep"`

### Task 9: Parallel per-target lanes with a small concurrency cap

**Classification:** high-risk (concurrency)
**Estimated implement time:** ~5 min
**Parallelizable with:** 2, 5, 6, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs` (`RetryPendingMessagesAsync`), `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptions.cs`
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs` (append)

Depends on Task 8. Removes head-of-line blocking *across* targets (one dead external system no longer delays notification forwarding to a healthy central). Ordering **within** a `(category, target)` lane stays strictly sequential — per-target FIFO is preserved.

1. Failing test:

```csharp
[Fact]
public async Task RetrySweep_SlowTarget_DoesNotBlockOtherTargets()
{
    var service = CreateService(); // options.SweepTargetParallelism default 4
    var slowGate = new TaskCompletionSource();
    var healthyDelivered = new TaskCompletionSource();
    service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, async msg =>
    {
        if (msg.Target == "slow") { await slowGate.Task; return true; }
        healthyDelivered.TrySetResult();
        return true;
    });
    await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "slow", "{}",
        attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);
    await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "healthy", "{}",
        attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero);

    var sweep = service.RetryPendingMessagesAsync();
    // Healthy lane completes while the slow lane is still blocked — pre-fix this
    // times out because delivery is strictly serial across targets.
    await healthyDelivered.Task.WaitAsync(TimeSpan.FromSeconds(5));
    slowGate.SetResult();
    await sweep;
}

[Fact]
public async Task RetrySweep_WithinTargetLane_StaysSequentialOldestFirst()
{
    var service = CreateService();
    var order = new System.Collections.Concurrent.ConcurrentQueue<string>();
    service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, msg =>
    {
        order.Enqueue(msg.Id); return Task.FromResult(true);
    });
    for (var i = 0; i < 3; i++)
        await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}",
            attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero, messageId: $"m{i}");
    await service.RetryPendingMessagesAsync();
    Assert.Equal(new[] { "m0", "m1", "m2" }, order.ToArray());
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "SlowTarget_DoesNotBlock|StaysSequentialOldestFirst"` — expect FAIL (first test).
3. Implement:
   - Option: `/// <summary>Max (category,target) lanes delivered concurrently per sweep. 1 = legacy serial. Within a lane delivery stays sequential (per-target FIFO).</summary> public int SweepTargetParallelism { get; set; } = 4;`
   - In `RetryPendingMessagesAsync`, replace the flat loop (keeping Task 8's short-circuit *inside* each lane):

```csharp
            var lanes = messages
                .GroupBy(m => (m.Category, m.Target))
                .Select(g => g.ToList())
                .ToList();

            using var laneCap = new SemaphoreSlim(Math.Max(1, _options.SweepTargetParallelism));
            var laneTasks = lanes.Select(async lane =>
            {
                await laneCap.WaitAsync();
                try
                {
                    foreach (var message in lane)
                    {
                        var outcome = await RetryMessageAsync(message);
                        if (outcome == RetryOutcome.TransientFailure)
                            return; // short-circuit the rest of this lane this sweep
                    }
                }
                finally { laneCap.Release(); }
            }).ToList();
            await Task.WhenAll(laneTasks);
```

     (`_bufferedCount` uses `Interlocked` and each storage call opens its own connection, so lanes are safe to run concurrently; the `failedTargets` set from Task 8 collapses into the per-lane `return`.)
4. Run the filter + full StoreAndForward.Tests project (watch for timing-sensitive existing sweep tests).
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests && git commit -m "perf(store-and-forward): parallel per-target sweep lanes (cap 4), sequential within lane"`

### Task 10: Ordered, observable replication dispatch

**Classification:** high-risk (replication ordering)
**Estimated implement time:** ~4 min
**Parallelizable with:** 1, 2, 5, 6, 7, 14
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs` (`FireAndForget`, lines 134-149), `src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs` (add counter), `src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs` (`SendToPeer`, lines 139-149)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs` (append)

`Task.Run` bought nothing (the handler is a non-blocking Akka `Tell` returning `Task.CompletedTask`) and cost ordering: Add/Remove pairs invert on the thread pool. Invoking inline preserves ordering by construction; Akka then guarantees per-sender/receiver order.

1. Failing test:

```csharp
[Fact]
public void ReplicationOperations_AreDispatchedInIssueOrder()
{
    var seen = new List<(ReplicationOperationType, string)>();
    var service = new ReplicationService(new StoreAndForwardOptions(), NullLogger<ReplicationService>.Instance);
    service.SetReplicationHandler(op =>
    {
        seen.Add((op.OperationType, op.MessageId));
        return Task.CompletedTask;
    });

    for (var i = 0; i < 200; i++)
    {
        service.ReplicateEnqueue(new StoreAndForwardMessage { Id = $"m{i}" });
        service.ReplicateRemove($"m{i}");
    }

    // Inline dispatch: by the time the calls return, every op was handed to the
    // handler, Add strictly before Remove per id. Pre-fix (Task.Run) this is
    // both racy-in-count and racy-in-order.
    Assert.Equal(400, seen.Count);
    for (var i = 0; i < 200; i++)
    {
        Assert.Equal((ReplicationOperationType.Add, $"m{i}"), seen[2 * i]);
        Assert.Equal((ReplicationOperationType.Remove, $"m{i}"), seen[2 * i + 1]);
    }
}

[Fact]
public void ReplicationHandlerThrow_IsSwallowed_AndLoggedAtWarning()
{
    var logger = new TestLogger<ReplicationService>(); // xunit-friendly capture, or use a Meter listener
    var service = new ReplicationService(new StoreAndForwardOptions(), logger);
    service.SetReplicationHandler(_ => throw new InvalidOperationException("peer gone"));
    service.ReplicateRemove("m1"); // must not throw
    Assert.Contains(logger.Entries, e => e.Level == LogLevel.Warning);
}
```

   (If no `TestLogger` helper exists in the test project, add a minimal `List<(LogLevel, string)>`-capturing `ILogger<T>` to the test file.)
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "ReplicationOperations_AreDispatchedInIssueOrder|ReplicationHandlerThrow"` — expect FAIL.
3. Implement:
   - `ScadaBridgeTelemetry`: add alongside the existing counters:

```csharp
    private static readonly Counter<long> _replicationFailures =
        Meter.CreateCounter<long>("scadabridge.store_and_forward.replication.failures", unit: "1",
            description: "S&F buffer replication operations that failed to dispatch/deliver to the peer node");
    /// <summary>Records one failed S&F replication dispatch.</summary>
    public static void RecordReplicationFailure() => _replicationFailures.Add(1);
```

   - `ReplicationService.FireAndForget` (rename to `DispatchInline`, update the three callers or keep the name with a corrected doc comment):

```csharp
    private void FireAndForget(ReplicationOperation operation)
    {
        // Invoked inline, NOT via Task.Run: the handler is a non-blocking Akka
        // Tell, and thread-pool hand-off destroyed Add/Remove ordering for the
        // same message id. Inline invocation preserves issue order; Akka's
        // per-sender/receiver guarantee preserves it on the wire.
        try
        {
            var task = _replicationHandler!.Invoke(operation);
            if (!task.IsCompletedSuccessfully)
            {
                task.ContinueWith(t =>
                    {
                        ScadaBridgeTelemetry.RecordReplicationFailure();
                        _logger.LogWarning(t.Exception,
                            "Replication of {OpType} for message {MessageId} failed (best-effort); standby buffer may be diverging",
                            operation.OperationType, operation.MessageId);
                    },
                    TaskContinuationOptions.OnlyOnFaulted);
            }
        }
        catch (Exception ex)
        {
            ScadaBridgeTelemetry.RecordReplicationFailure();
            _logger.LogWarning(ex,
                "Replication of {OpType} for message {MessageId} failed (best-effort); standby buffer may be diverging",
                operation.OperationType, operation.MessageId);
        }
    }
```

   - `SiteReplicationActor.SendToPeer` no-peer drop: raise `LogDebug` → `LogWarning` and call `ScadaBridgeTelemetry.RecordReplicationFailure()` (a dropped op is a lost delta; when the peer is legitimately absent — single-node dev — the warning is rate-tolerable at one per op; acceptable per review).
4. Run the filter + `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests` + `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter SiteReplication`.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs && git commit -m "fix(store-and-forward): inline ordered replication dispatch; Warning + counter on replication failures"`

### Task 11: Upsert-based standby applies for Add/Park/Requeue

**Classification:** standard
**Estimated implement time:** ~4 min
**Parallelizable with:** 2, 5, 6, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs` (add `UpsertMessageAsync`), `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs` (`ApplyReplicatedOperationAsync`, lines 107-132)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs` (append)

Depends on Task 10 (same `ReplicationService.cs`) and Task 7 (same storage file). A Park/Requeue whose original Add was lost self-heals (the full message rides in the operation); a duplicate Add after Task 21's resync no longer violates the PK.

1. Failing tests:

```csharp
[Fact]
public async Task ApplyReplicatedPark_WhenAddWasLost_MaterializesTheParkedRow()
{
    var service = new ReplicationService(OptionsWithReplication(), NullLogger<ReplicationService>.Instance);
    var msg = NewMessage("lost-add-1"); // never Added on this (standby) storage
    msg.Status = StoreAndForwardMessageStatus.Parked;

    await service.ApplyReplicatedOperationAsync(
        new ReplicationOperation(ReplicationOperationType.Park, msg.Id, msg), _storage);

    var row = await _storage.GetMessageByIdAsync("lost-add-1");
    Assert.NotNull(row); // pre-fix: blind UPDATE affected 0 rows, row is gone forever
    Assert.Equal(StoreAndForwardMessageStatus.Parked, row!.Status);
}

[Fact]
public async Task ApplyReplicatedAdd_Twice_IsIdempotent_NewestWins()
{
    var service = new ReplicationService(OptionsWithReplication(), NullLogger<ReplicationService>.Instance);
    var msg = NewMessage("dup-add");
    await service.ApplyReplicatedOperationAsync(
        new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage);
    msg.RetryCount = 3;
    await service.ApplyReplicatedOperationAsync( // pre-fix: SqliteException PK violation
        new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage);

    Assert.Equal(3, (await _storage.GetMessageByIdAsync("dup-add"))!.RetryCount);
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "ApplyReplicatedPark_WhenAddWasLost|ApplyReplicatedAdd_Twice"` — expect FAIL.
3. Implement:
   - `StoreAndForwardStorage.UpsertMessageAsync(StoreAndForwardMessage message)`: same column list/parameters as `EnqueueAsync` but `INSERT INTO sf_messages (...) VALUES (...) ON CONFLICT(id) DO UPDATE SET category=excluded.category, target=excluded.target, payload_json=excluded.payload_json, retry_count=excluded.retry_count, max_retries=excluded.max_retries, retry_interval_ms=excluded.retry_interval_ms, last_attempt_at=excluded.last_attempt_at, status=excluded.status, last_error=excluded.last_error, origin_instance=excluded.origin_instance, execution_id=excluded.execution_id, source_script=excluded.source_script, parent_execution_id=excluded.parent_execution_id` (keep the original `created_at` — do not overwrite it, it orders the sweep).
   - `ApplyReplicatedOperationAsync`: `Add`, `Park`, and `Requeue` all call `storage.UpsertMessageAsync(operation.Message)` (after the existing Status/RetryCount mutations for Park/Requeue). `Remove` unchanged. Update the doc comment: replicated applies are upserts so a lost Add self-heals from any later full-message operation.
4. Run the filter + full project.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests && git commit -m "fix(store-and-forward): standby applies Add/Park/Requeue as upserts so a lost Add self-heals"`

### Task 12: `deferToSweep` enqueue mode + immediate sweep kick

**Classification:** high-risk (delivery-path semantics)
**Estimated implement time:** ~5 min
**Parallelizable with:** 2, 5, 6, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs` (`EnqueueAsync` signature + tail, lines 467-546; new `TriggerSweep`)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs` (append)

Depends on Tasks 8/9 (same file). Groundwork for Task 13: `Notify.Send` must never run the 30 s central Ask inline on the script's await. `deferToSweep: true` buffers without invoking the handler and **without** stamping `LastAttemptAt` (row is due immediately), then kicks a background sweep so the healthy-path latency is milliseconds, not one timer interval.

1. Failing tests:

```csharp
[Fact]
public async Task Enqueue_DeferToSweep_DoesNotInvokeHandlerInline_AndRowIsDueImmediately()
{
    var service = CreateService();
    var inlineInvoked = false;
    service.RegisterDeliveryHandler(StoreAndForwardCategory.Notification,
        _ => { inlineInvoked = true; return Task.FromResult(true); });

    var result = await service.EnqueueAsync(StoreAndForwardCategory.Notification, "ops", "{}",
        deferToSweep: true, messageId: "n1");

    Assert.False(inlineInvoked);
    Assert.True(result.WasBuffered);
    var row = await service.GetMessageByIdAsync("n1");
    Assert.Null(row!.LastAttemptAt); // due on the very next sweep, not after RetryInterval
}

[Fact]
public async Task Enqueue_DeferToSweep_AfterStart_KicksAnImmediateSweep()
{
    var service = CreateService(retryTimerInterval: TimeSpan.FromHours(1)); // timer will never fire in-test
    var delivered = new TaskCompletionSource();
    service.RegisterDeliveryHandler(StoreAndForwardCategory.Notification,
        _ => { delivered.TrySetResult(); return Task.FromResult(true); });
    await service.StartAsync();
    try
    {
        await service.EnqueueAsync(StoreAndForwardCategory.Notification, "ops", "{}", deferToSweep: true);
        await delivered.Task.WaitAsync(TimeSpan.FromSeconds(5)); // pre-fix: times out
    }
    finally { await service.StopAsync(); }
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "Enqueue_DeferToSweep"` — expect FAIL (no parameter).
3. Implement:
   - Add parameter `bool deferToSweep = false` to `EnqueueAsync` (after `attemptImmediateDelivery`), XML-doc: *"When true, the handler is never invoked inline — the message is buffered due-immediately (`LastAttemptAt` stays null) and a background sweep is kicked. Use for callers on latency-sensitive threads (script dispatchers) whose delivery involves a long Ask timeout. Unlike `attemptImmediateDelivery: false`, no delivery attempt is presumed to have been made."*
   - Guard: `if (deferToSweep) { await BufferAsync(message); RaiseActivity("Queued", category, $"Deferred to sweep: {target}"); TriggerSweep(); return new StoreAndForwardResult(true, message.Id, true); }` placed **before** the immediate-delivery block. (`LastAttemptAt` intentionally not set.)
   - Add:

```csharp
    /// <summary>
    /// Kicks a background retry sweep now (fire-and-forget). No-op before
    /// StartAsync (storage may be uninitialized). Overlap-safe: the sweep's
    /// _retryInProgress CAS makes a concurrent kick a cheap no-op.
    /// </summary>
    public void TriggerSweep()
    {
        if (_retryTimer == null) return;
        Volatile.Write(ref _sweepTask, RetryPendingMessagesAsync());
    }
```

   - `CreateService` helper: add optional `retryTimerInterval` parameter feeding `StoreAndForwardOptions.RetryTimerInterval` if not already supported.
4. Run the filter + full project.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs && git commit -m "feat(store-and-forward): deferToSweep enqueue mode with immediate sweep kick for latency-sensitive callers"`

### Task 13: Unstrand and unblock `Notify.Send` (maxRetries: 0 + deferToSweep) + docs

**Classification:** high-risk (script-facing semantics)
**Estimated implement time:** ~5 min
**Parallelizable with:** 5, 6, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs` (the `EnqueueAsync` call at lines ~2197-2203)
- Modify: `docs/requirements/Component-StoreAndForward.md` (notification-delivery paragraph, ~line 52 region), `CLAUDE.md` (the "`Notify.Send` is async — returns … immediately" bullet: note it is now enqueue-only + swept, unbounded retries)
- Test: `tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/NotifyHelperTests.cs` (append)

Depends on Task 12. Two changes at the single notification producer: `maxRetries: 0` (the documented no-limit escape hatch — a central maintenance window can no longer strand notifications into per-message operator unparking) and `deferToSweep: true` (a `Notify.Send` during a central outage returns in milliseconds instead of blocking a scarce script thread for the 30 s `NotificationForwardTimeout`).

1. Failing test (the `NotifyHelperTests` harness constructs a real `StoreAndForwardService` at line 45):

```csharp
[Fact]
public async Task NotifySend_BuffersUnbounded_AndNeverInvokesForwarderInline()
{
    var inlineInvoked = false;
    _saf.RegisterDeliveryHandler(StoreAndForwardCategory.Notification,
        _ => { inlineInvoked = true; throw new TimeoutException("central down"); });

    var notificationId = await SendViaNotifyHelper("ops-list", "hello"); // existing helper pattern

    Assert.False(inlineInvoked); // pre-fix: Send ran the forwarder Ask inline
    var row = await _saf.GetMessageByIdAsync(notificationId);
    Assert.NotNull(row);
    Assert.Equal(0, row!.MaxRetries); // pre-fix: 50 → parked after ~25 min of outage
    Assert.Equal(StoreAndForwardMessageStatus.Pending, row.Status);
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter NotifySend_BuffersUnbounded` — expect FAIL.
3. Implement — the `ScriptRuntimeContext` call becomes:

```csharp
            await _storeAndForward.EnqueueAsync(
                StoreAndForwardCategory.Notification,
                target: _listName,
                payloadJson: payloadJson,
                originInstanceName: _instanceName,
                // 0 = the documented "no limit" escape hatch (StoreAndForward-015):
                // notifications are retried until central acks and are never parked
                // for retry exhaustion — a long central outage must not strand them
                // behind per-message operator unparking.
                maxRetries: 0,
                // Never run the forwarder's 30s central Ask inline on the script
                // thread: buffer due-immediately and kick the sweep. Send returns
                // in milliseconds whether central is up or down.
                deferToSweep: true,
                messageId: notificationId);
```

4. Update the docs: `Component-StoreAndForward.md` — rewrite the "bounded by DefaultMaxRetries … will park the buffered notification" passage: `Notify.Send` now enqueues with `maxRetries: 0`; notifications never park for retry exhaustion (corrupt payloads still park — Task 14); `Notify.Send` is enqueue-only (no inline forward attempt) with an immediate sweep kick, so its worst-case latency is the local SQLite insert. `CLAUDE.md`: adjust the `Notify.Send` bullet accordingly.
5. Run the filter + `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter Notify` — expect PASS (fix any existing Notify tests that asserted the inline-attempt behavior).
6. Commit: `git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRuntimeContext.cs docs/requirements/Component-StoreAndForward.md CLAUDE.md tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Scripts/NotifyHelperTests.cs && git commit -m "fix(notifications): Notify.Send enqueues unbounded (maxRetries 0) and defers to sweep — no 30s script-thread block, no stranding"`

### Task 14: Park (don't silently drop) corrupt buffered notifications

**Classification:** standard
**Estimated implement time:** ~4 min
**Parallelizable with:** 1, 2, 3, 5, 7, 10, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/NotificationForwarder.cs` (lines 77-106)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/NotificationForwarderTests.cs` (modify lines 201-255 — the two StoreAndForward-018 regression tests invert)
- Modify: `docs/requirements/Component-StoreAndForward.md` (note superseding StoreAndForward-018)

**Deliberate reversal of StoreAndForward-018.** That decision discarded corrupt payloads to honour "notifications do not park". The doc itself now frames "do not park" as happy-path intent, and Task 13 makes retry-exhaustion parking impossible (`maxRetries: 0`) — so parking is exactly the designed home for an *undeliverable-but-preserved* payload: the operator gets a parked row with the payload for forensics instead of a silent site-log-only vanish that central reports as delivered.

1. Rewrite the two tests (rename; flip assertions):

```csharp
[Fact]
public async Task Deliver_CorruptJsonPayload_ReturnsFalse_ParksForForensics_AndDoesNotForward()
{
    // Supersedes StoreAndForward-018 (discard-on-corrupt). Returning false is the
    // handler contract's permanent-failure signal: the engine parks the row, which
    // preserves the payload for operator forensics. Retrying a corrupt payload is
    // pointless; deleting it silently (and reporting Delivered) loses data with no
    // parked row, no central Notifications row, and no audit trail.
    var centralProbe = CreateTestProbe();
    var forwarder = new NotificationForwarder(centralProbe.Ref, "site-7", ForwardTimeout);
    var corrupt = new StoreAndForwardMessage
    {
        Id = "msg-corrupt", Category = StoreAndForwardCategory.Notification,
        Target = "Operators", PayloadJson = "{not-valid-json", OriginInstanceName = "Plant.Pump3",
    };

    Assert.False(await forwarder.DeliverAsync(corrupt));
    centralProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
}
```

   (Mirror for the `"null"` payload test.)
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "Deliver_CorruptJsonPayload|Deliver_NullDeserializedPayload"` — expect FAIL.
3. Implement in `DeliverAsync`:

```csharp
        if (!TryBuildSubmit(message, out var submit))
        {
            _logger.LogWarning(
                "Parking corrupt buffered notification {NotificationId} (payload is not deserialisable as NotificationSubmit); " +
                "the row is preserved for operator forensics. Payload preview: {PayloadPreview}",
                message.Id,
                PreviewPayload(message.PayloadJson));
            // false = permanent failure by the delivery-handler contract → the
            // engine parks the row (payload preserved, operator can inspect or
            // discard). Supersedes StoreAndForward-018's silent discard, which
            // reported the notification as delivered while losing it entirely.
            return false;
        }
```

   Update the class XML doc list (`true`/throws/`false` outcomes) to include the park case.
4. Update `Component-StoreAndForward.md`: corrupt buffered notifications park (StoreAndForward-018 superseded); note the two parking causes for notifications are now *corrupt payload only* (retry exhaustion is off via `maxRetries: 0`).
5. Run the filter + full project.
6. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/NotificationForwarder.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/NotificationForwarderTests.cs docs/requirements/Component-StoreAndForward.md && git commit -m "fix(notifications): park corrupt buffered notification payloads instead of silently discarding as delivered (supersedes StoreAndForward-018)"`

### Task 15: Wire-format pin tests for cross-cluster contracts

**Classification:** standard
**Estimated implement time:** ~5 min
**Parallelizable with:** 1, 2, 3, 5, 7, 10, 14
**Files:**
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/WireSerializationPinTests.cs` (create), `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationWireSerializationPinTests.cs` (create)

No production code change. The Akka default reflective-JSON wire format couples compatibility to *type and namespace identity*; nothing tests it (`MessageContractTests` asserts correlation-ids only). These tests pin (a) round-trip fidelity and (b) the fully-qualified type identity of the highest-traffic cross-cluster contracts, so a rename/move that would break rolling central↔site compat fails a test instead of production. The serializer *swap* (proto/explicit bindings) is **deferred** — it changes the wire format and needs a coordinated rolling-upgrade plan.

1. Write the tests (they pass immediately if the wire behavior is sane — the value is the pin; expect PASS on first run, and that's fine: the "failing" phase here is meaningless, this is characterization):

```csharp
// tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/WireSerializationPinTests.cs
public class WireSerializationPinTests : TestKit
{
    private T RoundTrip<T>(T message)
    {
        var serialization = Sys.Serialization;
        var serializer = serialization.FindSerializerFor(message);
        var bytes = serializer.ToBinary(message);
        return (T)serialization.Deserialize(bytes, serializer.Identifier,
            message!.GetType().TypeQualifiedName());
    }

    [Fact]
    public void HeartbeatMessage_RoundTripsOnTheWire()
    {
        var original = new HeartbeatMessage("site-1", "host-a", true, DateTimeOffset.UtcNow);
        var back = RoundTrip(original);
        Assert.Equal(original, back);
    }

    [Fact]
    public void NotificationSubmit_RoundTripsOnTheWire() { /* construct with all fields non-default; assert field equality */ }

    [Fact]
    public void IngestAuditEventsCommand_RoundTripsOnTheWire() { /* one fully-populated AuditEvent entity */ }

    [Fact]
    public void SiteHealthReport_RoundTripsOnTheWire() { /* minimal + one dictionary entry per collection */ }

    // Type-identity pins: the wire embeds CLR type manifests; a rename/move of any
    // of these types breaks rolling cross-version central↔site communication.
    // If one of these assertions fails, you are making a wire-breaking change —
    // coordinate a full-fleet upgrade or add an explicit serialization binding.
    [Theory]
    [InlineData(typeof(HeartbeatMessage), "ZB.MOM.WW.ScadaBridge.Commons.Messages.Health.HeartbeatMessage")]
    [InlineData(typeof(NotificationSubmit), "ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification.NotificationSubmit")]
    [InlineData(typeof(IngestAuditEventsCommand), "ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit.IngestAuditEventsCommand")]
    public void CrossClusterContract_TypeIdentity_IsPinned(Type type, string expectedFullName) =>
        Assert.Equal(expectedFullName, type.FullName);
}
```

   And in `ReplicationWireSerializationPinTests.cs` (StoreAndForward.Tests, which references the S&F assembly): round-trip `ReplicationOperation` carrying a fully-populated `StoreAndForwardMessage` (every property non-default, including `ExecutionId`/`ParentExecutionId`) and pin its type identity.
2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter WireSerializationPinTests` and `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter ReplicationWireSerializationPinTests` — expect PASS. If any round-trip FAILS, that is a live wire-fidelity bug: fix the message type (missing ctor for deserialization, get-only property), not the test.
3. Commit: `git add tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/WireSerializationPinTests.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationWireSerializationPinTests.cs && git commit -m "test(communication): pin wire round-trip + type identity of cross-cluster message contracts"`

### Task 16: Replicate heartbeat marks to the peer central node

**Classification:** small
**Estimated implement time:** ~4 min
**Parallelizable with:** 3, 5, 7, 10, 14
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs` (`HandleHeartbeat` lines 349-353; new record + Receive)
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs` (append)

Depends on Task 2 (same file). Mirrors the existing `SiteHealthReportReplica` pattern exactly; `MarkHeartbeat` is idempotent (timestamp overwrite), so self-delivery via pub-sub is harmless.

1. Failing test (follow the existing DI-stub pattern that substitutes `ICentralHealthAggregator`):

```csharp
[Fact]
public void HeartbeatReplica_MarksLocalAggregator_WithoutRebroadcast()
{
    var aggregator = Substitute.For<ICentralHealthAggregator>();
    var actor = CreateActorWith(aggregator); // existing helper pattern
    var ts = DateTimeOffset.UtcNow;

    actor.Tell(new SiteHeartbeatReplica(new HeartbeatMessage("site-1", "host-a", true, ts)));

    AwaitAssert(() => aggregator.Received(1).MarkHeartbeat("site-1", ts));
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter HeartbeatReplica_MarksLocalAggregator` — expect FAIL (record missing).
3. Implement:
   - Record (bottom of file, next to the other registration records): `/// <summary>Peer-replication envelope for a site heartbeat, fanned out over the same DistributedPubSub topic as SiteHealthReportReplica so both central aggregators mark heartbeats regardless of which node the site's ClusterClient delivered to.</summary> public sealed record SiteHeartbeatReplica(HeartbeatMessage Heartbeat);`
   - Constructor: `Receive<SiteHeartbeatReplica>(r => MarkHeartbeatLocally(r.Heartbeat));`
   - `HandleHeartbeat`: extract `MarkHeartbeatLocally(heartbeat)` (the existing two lines), then publish `new Publish(HealthReportTopic, new SiteHeartbeatReplica(heartbeat))` inside the same try/catch-no-op-for-TestKit pattern used by `HandleSiteHealthReport`.
4. Run the filter + full project.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs && git commit -m "fix(communication): replicate heartbeat marks to the peer central node (closes post-failover offline blind window)"`

### Task 17: S&F SQLite hygiene — WAL + busy_timeout

**Classification:** standard
**Estimated implement time:** ~4 min
**Parallelizable with:** 2, 5, 6, 13, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs` (`InitializeAsync` line 32-79; new `OpenConnectionAsync` helper; replace every `new SqliteConnection` + `OpenAsync` pair — ~12 sites)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs` (append)

Depends on Task 11 (same file — includes the new `UpsertMessageAsync`).

1. Failing test:

```csharp
[Fact]
public async Task Initialize_EnablesWalJournalMode_OnFileDatabase()
{
    var path = Path.Combine(_tempDir, "wal-test.db");
    var storage = new StoreAndForwardStorage($"Data Source={path}", NullLogger<StoreAndForwardStorage>.Instance);
    await storage.InitializeAsync();

    await using var conn = new SqliteConnection($"Data Source={path}");
    await conn.OpenAsync();
    await using var cmd = conn.CreateCommand();
    cmd.CommandText = "PRAGMA journal_mode";
    Assert.Equal("wal", (string)(await cmd.ExecuteScalarAsync())!);
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter Initialize_EnablesWalJournalMode` — expect FAIL (`delete`).
3. Implement:
   - `InitializeAsync`, right after `OpenAsync()`: execute `PRAGMA journal_mode=WAL;` (persistent, file-scoped; in-memory DBs return `memory` — harmless, don't assert).
   - Add and use everywhere:

```csharp
    /// <summary>
    /// Opens a connection with a 5s busy_timeout. Concurrent writers exist by
    /// design (script enqueues, the sweep's lanes, standby replication applies,
    /// central pull queries); with connection-per-operation the pragma must be
    /// set per open (it is per-connection, and Microsoft.Data.Sqlite pooling
    /// makes the extra statement cheap on a pooled physical connection).
    /// </summary>
    private async Task<SqliteConnection> OpenConnectionAsync()
    {
        var connection = new SqliteConnection(_connectionString);
        await connection.OpenAsync();
        await using var pragma = connection.CreateCommand();
        pragma.CommandText = "PRAGMA busy_timeout = 5000";
        await pragma.ExecuteNonQueryAsync();
        return connection;
    }
```

   - Mechanical replace in every method: `await using var connection = new SqliteConnection(_connectionString); await connection.OpenAsync();` → `await using var connection = await OpenConnectionAsync();`.
4. Run the filter + full project.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs && git commit -m "perf(store-and-forward): enable WAL + per-connection busy_timeout on the S&F SQLite store"`

### Task 18: Epoch-ms due-check (replace per-row julianday parsing)

**Classification:** standard
**Estimated implement time:** ~5 min
**Parallelizable with:** 2, 5, 6, 13, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs` (schema migration in `InitializeAsync`; writers `EnqueueAsync`/`UpdateMessageAsync`/`UpdateMessageIfStatusAsync`/`UpsertMessageAsync`/`RetryParkedMessageAsync`; due query lines 189-198)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs` (append)

Depends on Task 17 (same file). Additive column, existing `AddColumnIfMissingAsync` pattern; the ISO-8601 text column stays authoritative for reads/back-compat, the ms column drives the due predicate.

1. Failing tests:

```csharp
[Fact]
public async Task GetMessagesForRetry_UsesEpochMs_RespectsInterval()
{
    var msg = NewPendingMessage("due-check", retryIntervalMs: 60_000);
    msg.LastAttemptAt = DateTimeOffset.UtcNow.AddSeconds(-30); // not yet due
    await _storage.EnqueueAsync(msg);
    Assert.Empty(await _storage.GetMessagesForRetryAsync());

    msg.LastAttemptAt = DateTimeOffset.UtcNow.AddSeconds(-90); // due
    await _storage.UpdateMessageAsync(msg);
    Assert.Single(await _storage.GetMessagesForRetryAsync());
}

[Fact]
public async Task Initialize_BackfillsEpochMs_ForLegacyRows()
{
    await _storage.InitializeAsync();
    // Simulate a legacy row: text timestamp present, ms column NULL.
    await ExecRaw("INSERT INTO sf_messages (id, category, target, payload_json, created_at, last_attempt_at, status) " +
                  "VALUES ('legacy', 0, 't', '{}', @c, @l, 0)",
                  ("@c", DateTimeOffset.UtcNow.ToString("O")),
                  ("@l", DateTimeOffset.UtcNow.AddHours(-1).ToString("O")));
    await _storage.InitializeAsync(); // second init runs the backfill
    var ms = await ScalarRaw<long?>("SELECT last_attempt_at_ms FROM sf_messages WHERE id='legacy'");
    Assert.NotNull(ms);
    Assert.InRange(ms!.Value, DateTimeOffset.UtcNow.AddHours(-1).AddMinutes(-1).ToUnixTimeMilliseconds(),
                              DateTimeOffset.UtcNow.ToUnixTimeMilliseconds());
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "GetMessagesForRetry_UsesEpochMs|Initialize_BackfillsEpochMs"` — expect FAIL (column missing).
3. Implement:
   - `InitializeAsync`: `await AddColumnIfMissingAsync(connection, "last_attempt_at_ms", "INTEGER");` then a one-time backfill: `UPDATE sf_messages SET last_attempt_at_ms = CAST((julianday(last_attempt_at) - 2440587.5) * 86400000 AS INTEGER) WHERE last_attempt_at IS NOT NULL AND last_attempt_at_ms IS NULL` (the only remaining julianday use — runs once per legacy DB, then never matches). Add `CREATE INDEX IF NOT EXISTS idx_sf_messages_status_due ON sf_messages(status, last_attempt_at_ms);`.
   - Every writer that sets `@lastAttempt` also sets `@lastAttemptMs` (`message.LastAttemptAt?.ToUnixTimeMilliseconds()` or DBNull); `RetryParkedMessageAsync` sets `last_attempt_at_ms = NULL`.
   - Due query: `WHERE status = @pending AND (last_attempt_at_ms IS NULL OR retry_interval_ms = 0 OR (@nowMs - last_attempt_at_ms) >= retry_interval_ms)` with `cmd.Parameters.AddWithValue("@nowMs", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds())`.
4. Run the filter + full project (existing due-check tests must still pass).
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs && git commit -m "perf(store-and-forward): epoch-ms due-check with additive column, backfill, and (status, due) index"`

### Task 19: Decouple audit-observer notification from the sweep (ordered channel)

**Classification:** high-risk (concurrency; audit-event ordering)
**Estimated implement time:** ~5 min
**Parallelizable with:** 2, 5, 6, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs` (`StartAsync`/`StopAsync`; the three `NotifyCachedCallObserverAsync` await sites at lines ~620-626, 688-694, 711-717 and the transient site ~711)
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs` (append)

Depends on Task 12 (same file). A slow observer (SQLite audit write) must not stretch the sweep; a single-reader unbounded channel preserves per-operation event order.

1. Failing test:

```csharp
[Fact]
public async Task SlowObserver_DoesNotBlockTheSweep_AndEventsArriveInOrder()
{
    var gate = new TaskCompletionSource();
    var observed = new List<CachedCallAttemptOutcome>();
    var observer = new DelegatingObserver(async ctx =>
    {
        await gate.Task; // first call blocks until released
        lock (observed) observed.Add(ctx.Outcome);
    });
    var service = CreateServiceWithObserver(observer); // existing harness pattern
    await service.StartAsync();
    try
    {
        service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
            _ => Task.FromResult(true));
        await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}",
            attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero,
            messageId: TrackedOperationId.New().ToString());

        var sweep = service.RetryPendingMessagesAsync();
        await sweep.WaitAsync(TimeSpan.FromSeconds(5)); // pre-fix: blocks on gate → times out

        gate.SetResult();
        await AssertEventually(() => { lock (observed) return observed.Count == 1; });
    }
    finally { await service.StopAsync(); }
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter SlowObserver_DoesNotBlockTheSweep` — expect FAIL (timeout).
3. Implement:
   - Fields: `private readonly Channel<Func<Task>> _observerQueue = Channel.CreateUnbounded<Func<Task>>(new UnboundedChannelOptions { SingleReader = true }); private Task? _observerPump;`
   - `StartAsync`: `_observerPump = Task.Run(async () => { await foreach (var work in _observerQueue.Reader.ReadAllAsync()) { try { await work(); } catch (Exception ex) { _logger.LogWarning(ex, "Cached-call audit observer threw; ignored (best-effort per alog.md §7)"); } } });`
   - The three sweep call sites become non-awaiting posts: `PostObserverNotification(message, outcome, lastError, httpStatus, attemptStartUtc, durationMs);` where `PostObserverNotification` snapshots nothing extra (each branch finishes mutating `message` before posting; the sweep re-reads fresh row instances next tick) and does `_observerQueue.Writer.TryWrite(() => NotifyCachedCallObserverAsync(...));`.
   - `StopAsync` (before clearing the gauge): `_observerQueue.Writer.TryComplete(); if (_observerPump is not null) { try { await _observerPump.WaitAsync(SweepShutdownWaitTimeout); } catch (TimeoutException) { _logger.LogWarning("Audit-observer pump did not drain within {Timeout}", SweepShutdownWaitTimeout); } }` — note: after TryComplete a restarted service needs a fresh channel; make the channel a non-readonly field re-created in `StartAsync` if existing tests restart the same instance.
   - XML-doc on `NotifyCachedCallObserverAsync`: now invoked from the single-reader pump — latency-isolated from the sweep, order preserved per posting sequence.
4. Run the filter + full project (especially `CachedCallAttemptEmissionTests` — they may need `StartAsync`/`StopAsync` bracketing or an explicit pump-drain helper; keep their ordering assertions intact).
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs && git commit -m "perf(store-and-forward): post audit-observer notifications to an ordered single-reader channel, latency-isolating the sweep"`

### Task 20: Buffer resync storage primitives (`GetAllMessagesAsync` / `ReplaceAllAsync`)

**Classification:** standard
**Estimated implement time:** ~4 min
**Parallelizable with:** 2, 5, 6, 13, 14, 16, 19
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs`
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs` (append)

Depends on Task 18 (same file). Anti-entropy groundwork (report Underdeveloped #4).

1. Failing tests:

```csharp
[Fact]
public async Task GetAllMessages_ReturnsEveryStatus_OldestFirst_UpToLimit()
{
    await _storage.EnqueueAsync(NewMessage("p1", status: StoreAndForwardMessageStatus.Pending, createdAt: Base));
    await _storage.EnqueueAsync(NewMessage("k1", status: StoreAndForwardMessageStatus.Parked, createdAt: Base.AddSeconds(1)));
    await _storage.EnqueueAsync(NewMessage("p2", status: StoreAndForwardMessageStatus.Pending, createdAt: Base.AddSeconds(2)));

    var (all, truncated) = await _storage.GetAllMessagesAsync(limit: 2);
    Assert.Equal(new[] { "p1", "k1" }, all.Select(m => m.Id));
    Assert.True(truncated);
}

[Fact]
public async Task ReplaceAll_SwapsTheEntireBuffer_Atomically()
{
    await _storage.EnqueueAsync(NewMessage("stale"));
    await _storage.ReplaceAllAsync(new[] { NewMessage("fresh-1"), NewMessage("fresh-2") });

    Assert.Null(await _storage.GetMessageByIdAsync("stale"));
    Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-1"));
    Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-2"));
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter "GetAllMessages_ReturnsEveryStatus|ReplaceAll_SwapsTheEntireBuffer"` — expect FAIL.
3. Implement:
   - `GetAllMessagesAsync(int limit)`: the standard SELECT column list, no status filter, `ORDER BY created_at ASC LIMIT @limitPlusOne` — fetch `limit + 1`, return `(rows.Take(limit).ToList(), rows.Count > limit)`.
   - `ReplaceAllAsync(IReadOnlyList<StoreAndForwardMessage> messages)`: one transaction — `DELETE FROM sf_messages`, then insert each row (extract the parameter-binding block shared with `EnqueueAsync` into a private `BindMessageParameters(SqliteCommand, StoreAndForwardMessage)` helper), commit. XML-doc: standby-side anti-entropy apply; never call on an active node.
4. Run the filter + full project.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs && git commit -m "feat(store-and-forward): GetAllMessagesAsync/ReplaceAllAsync storage primitives for standby buffer resync"`

### Task 21: Peer-join S&F buffer resync (anti-entropy)

**Classification:** high-risk (cluster protocol)
**Estimated implement time:** ~5 min
**Parallelizable with:** 2, 5, 6, 14, 16
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs` (new messages, handlers, trigger in `TryTrackPeer`; ctor gains `Func<bool>? isActiveOverride = null` seam)
- Modify: `docs/requirements/Component-StoreAndForward.md` (Persistence section — document the resync)
- Test: `tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs` (append; follow the existing `SendToPeer`-override subclass pattern)

Depends on Tasks 10 (same file) and 20 (primitives). Closes "a standby down for an hour rejoins and diverges forever": on tracking a peer, a **standby** requests a full-buffer snapshot; the **active** node answers with up to `MaxResyncRows` (10 000) rows; the standby replaces its buffer wholesale. Combined with Task 11's upserts, later replicated ops land cleanly on the resynced state.

1. Failing tests:

```csharp
[Fact]
public void StandbyTrackingPeer_SendsResyncRequest()
{
    var actor = CreateActor(isActive: () => false, sendToPeerCapture: out var sent);
    SimulatePeerUp(actor); // existing MemberUp-injection pattern
    AwaitAssert(() => Assert.Contains(sent, m => m is RequestSfBufferResync));
}

[Fact]
public void ActiveTrackingPeer_DoesNotRequestResync()
{
    var actor = CreateActor(isActive: () => true, sendToPeerCapture: out var sent);
    SimulatePeerUp(actor);
    ExpectNoMsg(200.Milliseconds());
    Assert.DoesNotContain(sent, m => m is RequestSfBufferResync);
}

[Fact]
public async Task ActiveNode_AnswersResyncRequest_WithFullBufferSnapshot()
{
    await _sfStorage.EnqueueAsync(NewMessage("m1"));
    var actor = CreateActor(isActive: () => true, sendToPeerCapture: out _);
    actor.Tell(new RequestSfBufferResync(), TestActor);
    var snapshot = ExpectMsg<SfBufferSnapshot>();
    Assert.Single(snapshot.Messages);
    Assert.False(snapshot.Truncated);
}

[Fact]
public async Task StandbyNode_AppliesSnapshot_ReplacingItsBuffer()
{
    await _sfStorage.EnqueueAsync(NewMessage("stale"));
    var actor = CreateActor(isActive: () => false, sendToPeerCapture: out _);
    actor.Tell(new SfBufferSnapshot(new List<StoreAndForwardMessage> { NewMessage("fresh") }, false));
    await AwaitAssertAsync(async () =>
    {
        Assert.Null(await _sfStorage.GetMessageByIdAsync("stale"));
        Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh"));
    });
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests --filter "ResyncRequest|ResyncRequest_WithFullBufferSnapshot|AppliesSnapshot_Replacing"` — expect FAIL (messages missing).
3. Implement in `SiteReplicationActor.cs`:
   - Messages (same file, public — they cross Akka remoting; POCO payload rides the default serializer, pinned by Task 15's pattern):

```csharp
/// <summary>Standby→active: request a full S&F buffer snapshot for anti-entropy resync (sent when a standby (re)tracks a peer).</summary>
public sealed record RequestSfBufferResync;

/// <summary>Active→standby: full-buffer snapshot; Truncated=true when the buffer exceeded MaxResyncRows (standby logs a Warning — divergence beyond the cap drains naturally).</summary>
public sealed record SfBufferSnapshot(List<StoreAndForwardMessage> Messages, bool Truncated);
```

   - Ctor: `Func<bool>? isActiveOverride = null` stored as `_isActive = isActiveOverride ?? DefaultIsActive;` where `DefaultIsActive` is the repo-standard leader+Up check via `_cluster` (same body as `SiteCommunicationActor.DefaultIsActiveCheck`; swap point for plan 01's helper).
   - `TryTrackPeer`: after setting `_peerAddress`, `if (!SafeIsActive()) SendToPeer(new RequestSfBufferResync());` (`SafeIsActive` wraps `_isActive()` in try/catch→false).
   - `Receive<RequestSfBufferResync>`: only when `SafeIsActive()`; capture `var replyTo = Sender;` then pipe `_sfStorage.GetAllMessagesAsync(MaxResyncRows)` (const `int MaxResyncRows = 10_000`) into `replyTo.Tell(new SfBufferSnapshot(rows, truncated))` via `Task.Run(...).PipeTo(replyTo, success: t => new SfBufferSnapshot(t.Rows, t.Truncated))`. Standby receiving a request ignores it (Debug log).
   - `Receive<SfBufferSnapshot>`: only when `!SafeIsActive()`; fire-and-forget `_sfStorage.ReplaceAllAsync(msg.Messages)` with a `ContinueWith` error log (matches `HandleApplyStoreAndForward`'s pattern); Warning when `msg.Truncated`.
4. Update `Component-StoreAndForward.md` Persistence: document the peer-join resync (trigger, direction, 10k cap, replace-all semantics, why upserts make trailing ops safe).
5. Run the filter + `dotnet test tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests`.
6. Commit: `git add src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs docs/requirements/Component-StoreAndForward.md tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests && git commit -m "feat(store-and-forward): peer-join full-buffer resync — standby anti-entropy for the S&F buffer"`

### Task 22: Conventions & observability cleanup batch (Lows)

**Classification:** small (batched Lows, concrete edits)
**Estimated implement time:** ~5 min
**Parallelizable with:** 4, 13, 17, 18, 20 (blocked only by 16 — shared `CentralCommunicationActor.cs`)
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs` (lines 243-258), `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/StreamRelayActor.cs` (lines 105-111), `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs` (lines 114-124), `src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs`, `src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs` (line 425), `docs/requirements/Component-Communication.md`
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsTests.cs` (append)

Five concrete edits:

1. **[Low: check-then-act]** `SiteStreamGrpcServer.cs:243-245` — add a comment above the `_activeStreams.Count >= _maxConcurrentStreams` check: `// Deliberately check-then-act: two concurrent subscribes at the boundary can both pass, overshooting the cap by at most (concurrent subscribers - 1). Bounded and benign — not worth a lock on this path.`
2. **[Low: DropOldest observability]** `SiteStreamGrpcServer.cs:256-257` — count evictions via the `itemDropped` overload:

```csharp
        long dropped = 0;
        var correlationId = request.CorrelationId;
        var channel = Channel.CreateBounded<SiteStreamEvent>(
            new BoundedChannelOptions(1000) { FullMode = BoundedChannelFullMode.DropOldest },
            _ =>
            {
                // Lossy-under-backpressure is the debug-view spec; make the real
                // loss visible: first eviction + every 500th thereafter.
                var n = Interlocked.Increment(ref dropped);
                if (n == 1 || n % 500 == 0)
                    _logger.LogWarning(
                        "Debug stream {CorrelationId} backpressure: {Dropped} oldest event(s) evicted so far",
                        correlationId, n);
            });
```

   And in `StreamRelayActor.WriteToChannel` (lines 105-111), replace the dead warning: `// TryWrite on a DropOldest bounded channel never returns false — eviction of the oldest item is counted by the channel's itemDropped callback (SiteStreamGrpcServer). This branch guards only a completed/closed channel.` and demote the log to Debug ("channel closed, dropping event").
3. **[Low: duplicated timeout constants]** `SiteStreamGrpcServer.cs:45` — change `private static readonly TimeSpan AuditIngestAskTimeout` to `internal static readonly` with a doc note that `CentralCommunicationActor` shares it; `CentralCommunicationActor.cs:114-124` — delete `DefaultAuditIngestAskTimeout` and initialize `_auditIngestAskTimeout = auditIngestAskTimeout ?? Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout;` (same assembly). Keep the fault-semantics divergence comment but append: `// Consolidating the two ingest transports (gRPC vs ClusterClient) is a scheduled follow-up — see arch review 02, Underdeveloped #1.`
4. **[Low: interval double duty]** `CommunicationOptions.cs` — add `/// <summary>Application-level site→central heartbeat cadence. Distinct from TransportHeartbeatInterval (the Akka.Remote failure-detector setting) — tuning the transport FD must not silently retune the health heartbeat.</summary> public TimeSpan ApplicationHeartbeatInterval { get; set; } = TimeSpan.FromSeconds(5);` and switch `SiteCommunicationActor.PreStart` (line 425) to `_options.ApplicationHeartbeatInterval`. Default equals the old effective value — no behavior change. Test: `[Fact] public void ApplicationHeartbeatInterval_DefaultsTo5s_IndependentOfTransport() { var o = new CommunicationOptions(); Assert.Equal(TimeSpan.FromSeconds(5), o.ApplicationHeartbeatInterval); o.TransportHeartbeatInterval = TimeSpan.FromSeconds(30); Assert.Equal(TimeSpan.FromSeconds(5), o.ApplicationHeartbeatInterval); }`
5. **[U7 doc]** `Component-Communication.md` (debug-stream section): add an explicit accepted-limitation note — debug sessions are process-local on the central node; a central restart/failover drops them with no `OnStreamTerminated` signal; the engineer re-establishes the session.

Steps: write the `CommunicationOptionsTests` test first → `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter ApplicationHeartbeatInterval` (FAIL) → apply all five edits → run full `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests` (PASS) → commit: `git add src/ZB.MOM.WW.ScadaBridge.Communication docs/requirements/Component-Communication.md tests/ZB.MOM.WW.ScadaBridge.Communication.Tests && git commit -m "chore(communication): low-severity cleanup — eviction counting, shared ingest timeout, split app heartbeat interval, doc notes"`

### Task 23: Oldest-parked-age health metric (parked-retention aging signal)

**Classification:** standard (additive message contract)
**Estimated implement time:** ~5 min
**Parallelizable with:** 2, 5, 6, 13, 16, 21
**Files:**
- Modify: `src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs` (add `GetOldestParkedCreatedAtAsync`), `src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs` (additive field), `src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthReportSender.cs` (lines ~105-118, next to the parked-count block) + the `ISiteHealthCollector` interface/implementation it calls
- Test: `tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs` (append)

Depends on Task 20 (same storage file). Parked rows persist until operator action by design; this gives operators the aging signal the review flagged as absent (a forgotten site's parked backlog now shows its age, not just its count). Central UI tile rendering is a display-only follow-up once the field flows.

1. Failing test:

```csharp
[Fact]
public async Task GetOldestParkedCreatedAt_ReturnsOldestParkedRow_OrNull()
{
    Assert.Null(await _storage.GetOldestParkedCreatedAtAsync());

    var old = NewMessage("old-parked", status: StoreAndForwardMessageStatus.Parked,
        createdAt: DateTimeOffset.UtcNow.AddDays(-3));
    var recent = NewMessage("new-parked", status: StoreAndForwardMessageStatus.Parked,
        createdAt: DateTimeOffset.UtcNow.AddHours(-1));
    var pending = NewMessage("pending", status: StoreAndForwardMessageStatus.Pending,
        createdAt: DateTimeOffset.UtcNow.AddDays(-9)); // must not count
    await _storage.EnqueueAsync(old);
    await _storage.EnqueueAsync(recent);
    await _storage.EnqueueAsync(pending);

    var oldest = await _storage.GetOldestParkedCreatedAtAsync();
    Assert.Equal(old.CreatedAt, oldest!.Value, TimeSpan.FromSeconds(1));
}
```

2. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests --filter GetOldestParkedCreatedAt` — expect FAIL.
3. Implement:
   - Storage: `SELECT MIN(created_at) FROM sf_messages WHERE status = @parked` → parse to `DateTimeOffset?` (null when no rows).
   - `SiteHealthReport`: append additive default parameter `double? OldestParkedMessageAgeSeconds = null` with a doc comment (age computed at report time on the site; null = no parked rows; additive-only contract evolution).
   - `ISiteHealthCollector`: add `SetOldestParkedMessageAgeSeconds(double? ageSeconds);` implement in the collector (store; include in the built report).
   - `HealthReportSender`: inside the existing parked-count `try` block: `var oldest = await _sfStorage.GetOldestParkedCreatedAtAsync(); _collector.SetOldestParkedMessageAgeSeconds(oldest.HasValue ? (DateTimeOffset.UtcNow - oldest.Value).TotalSeconds : null);`
4. Build the solution (`dotnet build ZB.MOM.WW.ScadaBridge.slnx` — the additive default param must not break any constructor caller) and run `dotnet test tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests` + `dotnet test tests/ZB.MOM.WW.ScadaBridge.Host.Tests`.
5. Commit: `git add src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs src/ZB.MOM.WW.ScadaBridge.HealthMonitoring tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests && git commit -m "feat(health): oldest-parked-message age on the site health report (parked-retention aging signal)"`

### Task 24: Pin the reconciliation cursor boundary contract (`>=` at the site read)

**Classification:** standard (test-first; possible one-operator fix)
**Estimated implement time:** ~5 min
**Parallelizable with:** 1, 3, 5, 7, 10, 14
**Files:**
- Test: `tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullAuditEventsTests.cs` (append; mirror in `SiteStreamPullSiteCallsTests.cs`)
- Possibly modify: the site-side since-cursor query invoked by `SiteStreamGrpcServer.PullAuditEvents` (`SiteStreamGrpcServer.cs:478-484` — follow the Ask/store call into the site audit store in `src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/`; same for `PullSiteCalls` at `:563-572` into the SiteRuntime tracking store)

Central advances `since_utc` from the last returned row's timestamp; rows *sharing* that boundary timestamp are only safe if the site-side read is **inclusive** (`>=`) — duplicates are absorbed by central's EventId dedup / upsert-on-newer, but an *exclusive* read silently loses same-timestamp rows forever.

1. Locate the site-side query behind `PullAuditEvents` (grep `since` in the store the gRPC handler Asks) and note the current operator.
2. Write the failing/pinning test in `SiteStreamPullAuditEventsTests` (its harness already stands up the server with a real/fake store — follow it):

```csharp
[Fact]
public async Task PullAuditEvents_BoundaryTimestamp_IsInclusive_EventIdDedupAbsorbsDuplicates()
{
    // Two rows share one OccurredAtUtc; a pull whose since_utc equals that
    // timestamp MUST return both (inclusive read). Central dedups on EventId,
    // so duplication is safe; exclusion is silent permanent loss.
    var boundary = DateTime.UtcNow;
    await SeedAuditRow(eventId: _e1, occurredAtUtc: boundary);
    await SeedAuditRow(eventId: _e2, occurredAtUtc: boundary);

    var reply = await PullSince(boundary);
    Assert.Equal(2, reply.Events.Count);
}
```

3. Run: `dotnet test tests/ZB.MOM.WW.ScadaBridge.Communication.Tests --filter PullAuditEvents_BoundaryTimestamp` — if the current operator is `>=` it PASSES (the pin is the deliverable); if it FAILS, change the site-side query operator from `>` to `>=` (one-character fix + doc comment: *"inclusive: central's cursor is the last returned timestamp; EventId idempotency absorbs the re-sent boundary rows, exclusivity would lose same-timestamp rows"*), rerun, expect PASS.
4. Mirror the same boundary test for `PullSiteCalls` (upsert-on-newer absorbs duplicates there).
5. Run both filters + full project.
6. Commit: `git add -A tests/ZB.MOM.WW.ScadaBridge.Communication.Tests src/ZB.MOM.WW.ScadaBridge.AuditLog src/ZB.MOM.WW.ScadaBridge.SiteRuntime && git commit -m "test(communication): pin inclusive (>=) reconciliation cursor reads at the site; EventId/upsert dedup absorbs boundary duplicates"`

---

## Dependencies on other plans

- **Plan 01 (Cluster/Host/Failover)** — owns the general active-node/singleton-host facility and the leader-vs-oldest correction. Task 3's gate is a `Func<bool>` seam and Task 4/Task 21 wire the interim repo-standard leader+Up check with an explicit swap-point comment; when plan 01's shared helper lands, replace those two lambda bodies (and `SiteCommunicationActor.DefaultIsActiveCheck`, which this plan defers to plan 01). Plan 01 also owns the **real two-node failover test rig**, which is the integration-level proof for Tasks 1-4 (kill -9 a node → assert single-node S&F delivery and ClusterClient survival); this plan's unit tests are the fast-feedback layer.
- **Plan 04 (Data/Audit backbone)** — owns central-side ingest idempotency and data-layer indexes; Task 24 only pins the *site-side* read contract and relies on plan-04-owned central dedup staying as-is.
- No other plan touches `src/ZB.MOM.WW.ScadaBridge.StoreAndForward` or `src/ZB.MOM.WW.ScadaBridge.Communication`; merge conflicts are expected only in `AkkaHostedService.cs` (plan 01 edits `BuildHocon`; Task 4 edits `RegisterSiteActorsAsync` — different regions).

## Execution order

**P0 (fix before the next production deployment):** Tasks 1 → 2 (Critical crash-loop), 3 → 4 (standby gate), 5 → 6 (shared-channel disposal), 7 → 8 (sweep collapse). These four pairs are mutually independent — run them as four parallel tracks.

**Critical path (file-serialized chains):**
- `StoreAndForwardService.cs`: 3 → 7 → 8 → 9 → 12 → 13 → 19
- `StoreAndForwardStorage.cs`: 7 → 11 → 17 → 18 → 20 → 23
- `ReplicationService.cs` / `SiteReplicationActor.cs`: 10 → 11, 10+20 → 21
- `CentralCommunicationActor.cs`: 1 → 2 → 16 → 22

**P1:** 10, 11, 12, 13, 14, 15, 16. **P2 (hardening/underdeveloped):** 9, 17, 18, 19, 20, 21, 22, 23, 24.

After each task: `dotnet build ZB.MOM.WW.ScadaBridge.slnx` + the named per-project test command. At plan end: full `dotnet test ZB.MOM.WW.ScadaBridge.slnx` **plus** `dotnet test tests/ZB.MOM.WW.ScadaBridge.CLI.Tests` (not in the slnx), and rebuild the docker cluster (`bash docker/deploy.sh`) since Host/actor runtime changed.