fix(comms): review findings — consumer-based debug orphan net, foreign-cancel triad, honest onConnected, served-row-exact retirement, full-rate reconcile

F1 (HIGH) DebugStreamBridgeActor: the 5-minute orphan net measured the MAILBOX (SetReceiveTimeout), and once stream events were correctly marked INotInfluenceReceiveTimeout nothing recurring reset it — the snapshot lands once and GrpcStreamStable once — so every healthy session self-terminated at ~6 min with a false "Site disconnected". Replaced with a periodic self-tick (ConsumerLivenessCheckInterval, 30s) over a consumer-last-seen stamp renewed only by DebugStreamConsumerAlive, which DebugStreamService Tells on a shared timer to every session still in its registry (holding a session there IS "a consumer is attached" — both the Blazor view and the SignalR hub release it on dispose/disconnect, and it works headless). Reverting the wrapper was rejected: it would restore the quiet-instance orphan bug. F2 (MED) SiteStreamGrpcClient: the RpcException(Cancelled) filter now requires cts.IsCancellationRequested. A peer-originated / channel-dispose Cancelled fired none of onError/onCompleted/onConnected, leaving SiteAlarmAggregatorActor with _streamDown=false forever (IsLive stuck true, reconcile reopen guard never fired). F3 (MED) SiteStreamGrpcClient: a header TIMEOUT is no longer reported as connected — that shape is exactly what an unreachable site produces, and it cleared _streamDown, consumed _seedOnConnect and launched a full snapshot fan-out at a dead site. AwaitHeadersAsync returns bool; the first received event is the fallback connected signal, fired at most once from headers OR first event. F4 (LOW-MED) SqliteAuditWriter.MarkReconciledUpToAsync: the blanket below-cursor UPDATE retired late-stamped inserts that were never served (then age-purged — silent loss). The flip is now bounded by insertion order: a Pending row retires only if its rowid is at or below the high-water mark of rows this instance has served from ReadPendingSinceAsync (clamped on purge, since SQLite reuses rowids); Forwarded rows are exempt (central ACKed them over the push path). At-least-once is unchanged. F5 (LOW) Documented the liveness dependency (a served row never covered by a later cursor stays Pending forever; PurgeExpiredAsync never purges Pending) in ISiteAuditQueue + Component-AuditLog.md, and added a cheap site-health signal: SiteAuditBacklogReporter logs a rate-limited warning when the existing oldest-pending metric exceeds 24h. F6 (MED) SiteAlarmAggregatorActor: _fanoutSinceLastTick was armed by the reconcile's OWN fan-out, so steady state ran fan-out→skip→fan-out→skip — one reconcile per 2x interval (120s), halving the not-reporting refresh and the alarm reconcile backstop. The skip is now armed only by connect/failover-driven seeds (initial, _seedOnConnect, and a re-seed queued behind one). Tests: Communication.Tests 691 passed (+13), AuditLog.Tests 382 passed (+5).
2026-08-14 23:52:25 -04:00
parent b1de9dfdd4
commit fd5e023d08
16 changed files with 1192 additions and 84 deletions
@@ -261,7 +261,15 @@ room is a compliance violation, not a self-healing behavior. To bound that
 growth in practice, the site emits a `SiteAuditBacklog` health metric (pending
 row count, oldest pending age, bytes on disk); crossing operator-configured
 thresholds surfaces a warning on the relevant site tile in the Health
-dashboard, mirroring the Store-and-Forward Engine's backlog metric.
+dashboard, mirroring the Store-and-Forward Engine's backlog metric, and
 `SiteAuditBacklogReporter` additionally logs a rate-limited warning once the
 oldest pending row passes 24 h.
 The same invariant carries a liveness dependency worth stating explicitly: the
 `Pending` floor clears only when central acknowledges the rows — a telemetry ack
 or a reconciliation cursor that covers them — never on age. See *Reconciliation
 pull* below for the served-row retirement rule that decides which rows can be
 acknowledged by a cursor at all.
 Central is the durable home. Site SQLite is a write-buffer with a forwarding
 guarantee.
@@ -367,6 +375,38 @@ it — a lagging drain is meant to surface as the stalled signal. The id tiebrea
 what makes that safe against a same-instant burst larger than one batch: the cursor
 advances on every tick even when the timestamp cannot.
 **Only rows that were actually served may retire.** `OccurredAtUtc` is stamped by the
 caller, so a row can be *inserted* after a batch was served yet carry a timestamp
 *below* central's (by then advanced) cursor — a back-dated stamp, a clock nudge, a
 write flushed late. A cursor flip that keyed on the timestamp alone retired exactly
 those rows: never served, never servable again (the keyset read has moved past them),
 and, being `Reconciled`, purged on age. That is silent audit loss, and it is a failure
 mode the pre-WP2.3 explicit id-set flip could not produce, so `MarkReconciledUpToAsync`
 carries a second, insertion-order bound:
 > A `Pending` row retires only if its insertion order is at or below the high-water
 > mark of rows this site node has actually served from `ReadPendingSinceAsync`
 > (SQLite `rowid`). A `Forwarded` row is exempt — central ACKED it over the telemetry
 > push path, which is proof independent of the pull.
 The bound is per-process, so after a site-node restart the first pull retires only
 `Forwarded` rows and the pull after it resumes normal retirement — conservative in the
 safe direction (retirement is delayed, never a row lost). The gRPC handler's ordering
 (retire, *then* read) is what keeps the bound from ever vouching for the batch it is
 about to serve.
 **Liveness note (accepted, documented).** A row that was served but never covered by a
 later cursor — central reconciliation stopped for good, or the bound reset over a
 restart before the next cursor arrived — stays `Pending` indefinitely, and the site
 retention purge never purges `Pending`. That is the hard `ForwardState` invariant
 working as intended (an unacknowledged row is not droppable), but it means the site
 store's floor depends on reconciliation actually running, not merely on the retention
 window elapsing. It is observable rather than silent: `GetBacklogStatsAsync` reports
 the pending count and oldest-pending instant on every site health report
 (`SiteAuditBacklog`), and `SiteAuditBacklogReporter` logs a rate-limited warning once
 the oldest pending row exceeds `StalePendingThreshold` (24 h) naming the drain and the
 reconciliation pull as the things to check.
 ### Central direct-write (central-originated events)
 Events originating at central never touch site SQLite. Inbound API writes one
@@ -70,7 +70,8 @@ Both central and site clusters. Each side has communication actors that handle m
 - **Bridge-session hardening (WP2.3):**
  - The pre-snapshot buffer is **bounded (20 000 events, drop-oldest)** and its evictions counted (`scadabridge.central.debug_view.presnapshot_dropped`). It was previously unbounded, so a session whose snapshot never arrived grew without limit on the CENTRAL node. Dropping the oldest is correct here: the snapshot that ends the buffering phase is authoritative for anything that old.
  - A **hard snapshot deadline** (`DebugStreamBridgeActor.SnapshotTimeout`, 60 s) fails the session if no `DebugViewSnapshot` arrives. Nothing else ended a session wedged in the buffering phase — a lost site reply raises no gRPC error.
-  - **Stream events no longer influence the orphan receive timeout.** The gRPC callback wraps each event in an envelope marked `INotInfluenceReceiveTimeout`, so a busy site can no longer keep an abandoned session alive forever by feeding it events. The 5-minute timeout now measures session/consumer liveness, which is what it was for.
+  - **The orphan net measures the CONSUMER, not the mailbox.** A session self-terminates after `DebugStreamBridgeActor.ConsumerIdleTimeout` (5 min) without a sign of life from its consumer, checked by a periodic self-tick (`ConsumerLivenessCheckInterval`, 30 s) against a consumer-last-seen stamp. The stamp is renewed only by `DebugStreamConsumerAlive`, which `DebugStreamService` Tells to every session still in its registry on a shared 30 s timer — holding a session there IS what "a consumer is attached" means, since both consumers (the Blazor debug view and the SignalR hub) release it on dispose/disconnect.
    This replaced an Akka `SetReceiveTimeout(5 min)`. That version measured the MAILBOX, which conflates site chatter with consumer liveness: a busy site kept an abandoned session alive forever. Marking stream events `INotInfluenceReceiveTimeout` fixed that but left the timeout with nothing recurring to reset it — the snapshot lands once and the stability tick once — so **every healthy session self-terminated ~6 min in and the operator was told "Site disconnected"**. Both failure directions are pinned by tests: streaming events with keepalives survive many windows; an orphaned session terminates while events keep arriving.
 ### 6.1 Aggregated Live Alarm Stream (Site → Central)
@@ -80,7 +81,8 @@ Delivered 2026-07-10 (`docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.m
 - **Central live cache** (`ISiteAlarmLiveCache`, singleton `SiteAlarmLiveCacheService`): a DI singleton on the active central node. For each site with ≥1 active viewer it runs ONE shared, **reference-counted** per-site aggregator (`SiteAlarmAggregatorActor`); the first `Subscribe(siteId, onChanged)` starts it, the last subscriber leaving stops it after a short **linger** to avoid re-seed thrash. `GetCurrentAlarms(siteId)` returns the current immutable snapshot; `IsLive(siteId)` reports whether the aggregator has seeded **and its site-wide stream is currently up**. Liveness rides every publish from the aggregator, so a stream that faults or ends gracefully drops `IsLive` immediately rather than leaving the page grafting a freezing snapshot over fresh poll data until the next reconcile (WP2.3).
 - **Seed-then-stream** (copied from `DebugStreamBridgeActor` ordering): open the `SubscribeSite` stream first (buffer live deltas), run the snapshot fan-out once via the existing `DebugViewSnapshot` path (bounded by `LiveAlarmCacheSeedConcurrency`), flush the buffer with **dedup by `(InstanceUniqueName, AlarmName, SourceReference)`**, then live pass-through. Placeholders are seeded from the snapshot and never expected on the live stream.
 - **Alarms-only seed (WP2.3)**: the seed/reconcile fan-out sets `DebugSnapshotRequest.AlarmsOnly` (wire: `DebugSnapshotRequestDto.alarms_only`, field 3, additive), so the site builds and ships only the alarm half of the snapshot. The fan-out discarded every attribute row anyway, and an instance's attribute surface dwarfs its alarm set. A pre-WP2.3 site ignores the flag and returns the full snapshot, which reads identically.
- **Failover & drift**: a re-seed runs **once per successful (re)connect**, not once per reconnect ATTEMPT — the connect signal is `SiteStreamGrpcClient.SubscribeSiteAsync`'s `onConnected` callback, raised when the site's response headers arrive (the site flushes them as soon as its relay is attached, so no event can be missed after it). Fanning a whole-site snapshot out per retry meant N snapshots against a site that was, by definition of the retry, unreachable. A periodic **reconcile snapshot** (default 60s, **jittered** by `LiveAlarmCacheReconcileJitterFraction` so aggregators started together do not stampede one boundary) remains the drift backstop, but it is **skipped when a fan-out already ran in that window** and **publishes only when the snapshot actually changed the cache** (a diff, not an unconditional viewer fan-out). Staleness stays bounded at two intervals: the skip consumes its flag, so the next tick always fans out. A fan-out that fails as a whole now retries on its own **backoff** timer (`reconnectDelay` doubling, capped at 8× the reconcile interval) instead of waiting a full interval. `[PERM]` (`docs/plans/2026-05-29-native-alarms-design.md`): the cache is **purely in-memory** — no EF entity/table/migration, no persisted central alarm store — so a new active node simply re-seeds from scratch.
+- **Failover & drift**: a re-seed runs **once per successful (re)connect**, not once per reconnect ATTEMPT — the connect signal is `SiteStreamGrpcClient.SubscribeSiteAsync`'s `onConnected` callback, raised when the site's response headers arrive (the site flushes them as soon as its relay is attached, so no event can be missed after it) or, for a peer that defers its headers, when the FIRST EVENT arrives — whichever comes first, and at most once. A header **timeout** is deliberately not a connect signal: an unreachable or wedged site produces exactly that shape, and reporting it as connected cleared `_streamDown`, consumed the pending re-seed and fanned a whole-site snapshot out at a site that never answered. Fanning a whole-site snapshot out per retry meant N snapshots against a site that was, by definition of the retry, unreachable. A periodic **reconcile snapshot** (default 60s, **jittered** by `LiveAlarmCacheReconcileJitterFraction` so aggregators started together do not stampede one boundary) remains the drift backstop, but it is **skipped when a CONNECT- OR FAILOVER-DRIVEN seed already ran in that window** and **publishes only when the snapshot actually changed the cache** (a diff, not an unconditional viewer fan-out). Staleness stays bounded at two intervals: the skip consumes its flag, so the next tick always fans out. A **tick's own fan-out never arms that skip** — when it did, steady state alternated fan-out/skip and the effective reconcile rate was one per TWO intervals (120 s by default), halving both the not-reporting refresh and the alarm reconcile backstop for no benefit: with no reconnect in play there is nothing duplicated to suppress. A fan-out that fails as a whole now retries on its own **backoff** timer (`reconnectDelay` doubling, capped at 8× the reconcile interval) instead of waiting a full interval. `[PERM]` (`docs/plans/2026-05-29-native-alarms-design.md`): the cache is **purely in-memory** — no EF entity/table/migration, no persisted central alarm store — so a new active node simply re-seeds from scratch.
 - **Stream terminations are a triad, and every ending hits exactly one leg.** `SiteStreamGrpcClient.ConsumeStreamAsync` classifies a stream's end as a fault (`onError`), a graceful server-side end (`onCompleted`), or our own teardown (neither). Only a cancellation **we asked for** takes the silent leg — the `RpcException(Cancelled)` filter is guarded by `cts.IsCancellationRequested`. A **foreign** `Cancelled` (the peer cancelled, or the channel was disposed underneath us) falls through to `onError`; unguarded, it fired none of the three, so the aggregator kept `_streamDown = false` — `IsLive` stuck true and the reconcile tick's reopen, which only runs on a stream known to be down, never fired.
 - **Options** (`Communication` section, `CommunicationOptions`; eagerly validated by `CommunicationOptionsValidator` / `ValidateOnStart`): `LiveAlarmCacheLinger` (default 30s), `LiveAlarmCacheReconcileInterval` (default 60s), `LiveAlarmCacheSeedConcurrency` (default 8), `LiveAlarmCacheMaxSubscribersPerSite` (default 200), `LiveAlarmCachePublishCoalesce` (default 250ms; `0` = publish per delta — legacy — batches an alarm storm into one snapshot copy + one viewer fan-out per window; arch review 02 round 2, N6), `LiveAlarmCacheReconcileJitterFraction` (default 0.2 = up to +20% per tick; `0` disables). Stream channel sizing lives alongside the other gRPC limits: `GrpcInstanceStreamChannelCapacity` (default 1000) and `GrpcSiteAlarmStreamChannelCapacity` (default 20 000).
 - **Telemetry** (`ScadaBridgeTelemetry` meter): observable gauge `scadabridge.site.alarm_cache.aggregators.active` (running per-site aggregators) and counter `scadabridge.site.alarm_cache.reconnects` (site-wide stream reconnects — a NodeA↔NodeB flip or reconcile-driven reopen; a sustained climb signals a flapping site link), and counter `scadabridge.site.alarm_cache.buffer_dropped` (deltas evicted from the aggregator's **bounded** 20 000-entry pre-seed buffer, drop-oldest — non-zero means a fan-out ran long enough for the delta storm behind it to exceed the cap; the fan-out's snapshot is authoritative for the evicted rows).
 - **Accepted limitations (arch review 02 round 2, N8):**
@@ -45,12 +45,27 @@ public sealed class SiteAuditBacklogReporter : IHostedService, IDisposable
    /// </summary>
    internal static readonly TimeSpan DefaultRefreshInterval = TimeSpan.FromSeconds(30);
    /// <summary>
    /// Age at which the oldest still-Pending row is called out in the log. A Pending row is
    /// one central has not acknowledged through EITHER path (telemetry push or reconciliation
    /// pull), and <c>PurgeExpiredAsync</c> deliberately never purges Pending — so the site
    /// store's floor depends on reconciliation actually running. Rows served by a pull but
    /// never covered by a later cursor sit in exactly this state, which is why the age is
    /// worth a signal rather than only a dashboard number. One day is comfortably longer than
    /// any normal drain/reconcile outage and well inside the ~7-day site retention window.
    /// </summary>
    internal static readonly TimeSpan StalePendingThreshold = TimeSpan.FromHours(24);
    /// <summary>How often the stale-pending warning may repeat (it is a standing condition).</summary>
    internal static readonly TimeSpan StalePendingWarnInterval = TimeSpan.FromHours(1);
    private readonly ISiteAuditQueue _queue;
    private readonly ISiteHealthCollector _collector;
    private readonly ILogger<SiteAuditBacklogReporter> _logger;
    private readonly TimeSpan _refreshInterval;
    private CancellationTokenSource? _cts;
    private Task? _loop;
    private DateTime _lastStalePendingWarnUtc = DateTime.MinValue;
    /// <summary>Initializes a new instance of <see cref="SiteAuditBacklogReporter"/>.</summary>
    /// <param name="queue">The site audit queue used to probe the backlog count.</param>
@@ -151,6 +166,7 @@ public sealed class SiteAuditBacklogReporter : IHostedService, IDisposable
        {
            var snapshot = await _queue.GetBacklogStatsAsync(ct).ConfigureAwait(false);
            _collector.UpdateSiteAuditBacklog(snapshot);
            WarnIfPendingIsStale(snapshot.OldestPendingUtc, snapshot.PendingCount);
        }
        catch (OperationCanceledException)
        {
@@ -166,6 +182,34 @@ public sealed class SiteAuditBacklogReporter : IHostedService, IDisposable
        }
    }
    /// <summary>
    /// Logs a rate-limited warning when the oldest still-Pending audit row is older than
    /// <see cref="StalePendingThreshold"/>. Pending rows are exempt from the retention purge
    /// by design, so a standing Pending backlog is the one site-store condition that does not
    /// self-heal on age — it clears only when central acknowledges the rows (telemetry ack or
    /// a reconciliation cursor that covers them). Internal so tests can drive it directly.
    /// </summary>
    /// <param name="oldestPendingUtc">Oldest pending row's occurrence instant, or null when none.</param>
    /// <param name="pendingCount">Number of rows currently pending.</param>
    internal void WarnIfPendingIsStale(DateTime? oldestPendingUtc, int pendingCount)
    {
        if (oldestPendingUtc is null) return;
        var age = DateTime.UtcNow - DateTime.SpecifyKind(oldestPendingUtc.Value, DateTimeKind.Utc);
        if (age < StalePendingThreshold) return;
        var now = DateTime.UtcNow;
        if (now - _lastStalePendingWarnUtc < StalePendingWarnInterval) return;
        _lastStalePendingWarnUtc = now;
        _logger.LogWarning(
            "Site audit backlog: oldest pending row is {AgeHours:F1}h old ({PendingCount} pending). " +
            "Pending rows are never purged on age, so this backlog only clears when central " +
            "acknowledges them — check the site→central audit telemetry drain and the central " +
            "reconciliation pull.",
            age.TotalHours, pendingCount);
    }
    /// <summary>Signals the polling loop to stop and waits for it to complete.</summary>
    /// <param name="ct">Cancellation token (not used; the internal CTS governs shutdown).</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
@@ -792,9 +792,14 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
                // ordering the query applies, so it is a strict "everything after this row".
                : "(fs.OccurredAtUtc > $since OR (fs.OccurredAtUtc = $since AND ae.EventId > $afterId))";
            // fs.rowid rides along as an 11th column purely to maintain
            // _maxServedRowId — the insertion-order high-water mark that bounds
            // MarkReconciledUpToAsync (see that method). It is never mapped onto
            // the returned AuditEvent; rowid stays a storage-layer concept.
            cmd.CommandText = $"""
                SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome,
-                       ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson
+                       ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson,
                       fs.rowid
                FROM   audit_event ae
                JOIN   audit_forward_state fs ON fs.EventId = ae.EventId
                WHERE  fs.ForwardState IN ($pending, $forwarded)
@@ -816,10 +821,47 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
            }
            cmd.Parameters.AddWithValue("$limit", batchSize);
-            return Task.FromResult(ReadRows(cmd, batchSize));
+            return Task.FromResult(ReadServedRows(cmd, batchSize));
        }
    }
    /// <summary>
    /// Highest <c>audit_forward_state.rowid</c> this instance has ever SERVED from
    /// <see cref="ReadPendingSinceAsync"/> — i.e. the insertion-order high-water mark of
    /// rows central could possibly have received through the pull path. Bounds
    /// <see cref="MarkReconciledUpToAsync"/> so a row inserted after that point can never be
    /// retired by a cursor that happens to sit above its timestamp. Monotonic (a later, smaller
    /// batch never lowers it) except for the purge clamp. Written under <c>_readLock</c> (the
    /// read path) and read under <c>_writeLock</c> (the flip/purge paths) — two different
    /// locks, so the accesses go through <see cref="Volatile"/> for visibility. A stale read
    /// can only make the bound smaller, i.e. more conservative, never unsafe.
    /// </summary>
    private long _maxServedRowId;
    /// <summary>
    /// Executes a reconciliation-pull read whose projection carries <c>fs.rowid</c> as an
    /// 11th column, materialising the canonical rows while advancing
    /// <see cref="_maxServedRowId"/>. Serving a row is what makes it eligible for
    /// cursor-proved retirement, so the two must happen in the same step.
    /// </summary>
    private IReadOnlyList<AuditEvent> ReadServedRows(SqliteCommand cmd, int capacityHint)
    {
        var rows = new List<AuditEvent>(Math.Min(capacityHint, 256));
        var maxRowId = Volatile.Read(ref _maxServedRowId);
        using (var reader = cmd.ExecuteReader())
        {
            while (reader.Read())
            {
                rows.Add(MapRow(reader));
                var rowId = reader.GetInt64(10);
                if (rowId > maxRowId) maxRowId = rowId;
            }
        }
        Volatile.Write(ref _maxServedRowId, maxRowId);
        return rows;
    }
    /// <summary>
    /// Normalises a wire-supplied event id to the exact textual form stored in
    /// <c>audit_event.EventId</c> so the keyset comparison is apples-to-apples. An
@@ -845,16 +887,40 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
                ? "fs.OccurredAtUtc < $since"
                : "(fs.OccurredAtUtc < $since OR (fs.OccurredAtUtc = $since AND fs.EventId <= $afterId))";
            // INSERTION-ORDER BOUND — "only rows we actually served can retire".
            //
            // The timestamp cursor alone is not enough. OccurredAtUtc is stamped by the
            // caller, so a row can be INSERTED after a batch was served yet carry a
            // timestamp BELOW central's (by then advanced) cursor — a late-stamped insert.
            // The blanket cursor UPDATE retired exactly those rows: never served, never
            // servable again (ReadPendingSinceAsync's keyset has moved past them), and,
            // being Reconciled, purged on age — silent audit loss, a failure mode the old
            // explicit id-set flip could not produce.
            //
            // fs.rowid is insertion order, so bounding the flip at the highest rowid this
            // instance has ever served restores the invariant exactly: rows inserted after
            // that point are out of the flip's reach no matter where the cursor sits. A
            // FORWARDED row is exempt from the bound — central ACKED it through the
            // telemetry push path, which is receipt-proof independent of the pull.
            //
            // Bound state is per-process: after a restart it is 0 until this instance serves
            // a batch, so the first pull retires only Forwarded rows and the pull after it
            // resumes normal retirement. Conservative in the safe direction (delays
            // retirement, never loses a row); the liveness note in MarkReconciledUpToAsync's
            // interface doc + Component-AuditLog.md covers what a permanently halted
            // reconciliation means for such rows.
            using var cmd = _connection.CreateCommand();
            cmd.CommandText = $"""
                UPDATE audit_forward_state AS fs
                SET    ForwardState = $reconciled
                WHERE  fs.ForwardState IN ($pending, $forwarded)
                  AND  (fs.ForwardState = $forwarded OR fs.rowid <= $maxServedRowId)
                  AND  {predicate};
                """;
            cmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString());
            cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
            cmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString());
            cmd.Parameters.AddWithValue("$maxServedRowId", Volatile.Read(ref _maxServedRowId));
            cmd.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString(
                "o", System.Globalization.CultureInfo.InvariantCulture));
            if (afterId is not null)
@@ -1063,6 +1129,22 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
                        dropCmd.ExecuteNonQuery();
                    }
                    // Clamp the served-rows high-water mark to what is still in the table.
                    // SQLite hands a fresh insert MAX(rowid)+1, so a purge that empties the
                    // sidecar (or removes its top rows) makes rowids REUSABLE — a stale-high
                    // _maxServedRowId would then vouch for rows nobody has served. Cheap:
                    // MAX(rowid) is an index-less O(1) lookup.
                    if (purged > 0)
                    {
                        using var maxCmd = _connection.CreateCommand();
                        maxCmd.Transaction = transaction;
                        maxCmd.CommandText = "SELECT IFNULL(MAX(rowid), 0) FROM audit_forward_state;";
                        var liveMax = Convert.ToInt64(maxCmd.ExecuteScalar(),
                            System.Globalization.CultureInfo.InvariantCulture);
                        if (liveMax < Volatile.Read(ref _maxServedRowId))
                            Volatile.Write(ref _maxServedRowId, liveMax);
                    }
                    transaction.Commit();
                }
                catch
@@ -143,6 +143,29 @@ public interface ISiteAuditQueue
    /// <paramref name="sinceUtc"/> are proven received; rows at the boundary instant are left
    /// alone. Idempotent; already-Reconciled rows are untouched.
    /// </para>
    /// <para>
    /// <b>Only-served-rows-retire.</b> The cursor is a timestamp, and
    /// <see cref="AuditEvent.OccurredAtUtc"/> is stamped by the caller, so a row can be
    /// INSERTED after a batch was served yet carry a timestamp below the (by then advanced)
    /// cursor. Implementations MUST NOT retire such a row: it was never served, the keyset
    /// read has moved past it, and retiring it would make it purgeable — silent audit loss.
    /// The SQLite implementation bounds the flip by insertion order (<c>rowid</c> high-water
    /// mark of rows it has served), exempting rows already
    /// <see cref="ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.AuditForwardState.Forwarded"/>
    /// because central ACKED those through the telemetry push path.
    /// </para>
    /// <para>
    /// <b>Liveness dependency (documented, accepted).</b> A row that WAS served but is never
    /// covered by a later cursor — central reconciliation stopped for good, or the site node
    /// restarted and its bound reset — stays
    /// <see cref="ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.AuditForwardState.Pending"/>
    /// indefinitely, and <see cref="PurgeExpiredAsync"/> never purges Pending. That is the
    /// retention invariant working as designed (an unacknowledged row is not droppable), but
    /// it means the site store's floor is bounded by reconciliation actually running. The
    /// backlog is observable: <see cref="GetBacklogStatsAsync"/> reports the pending count and
    /// oldest-pending instant on every site health report, and the site reporter logs a
    /// warning once the oldest pending row exceeds its stale threshold.
    /// </para>
    /// </remarks>
    /// <param name="sinceUtc">The cursor timestamp central has consumed up to (UTC).</param>
    /// <param name="afterId">The last consumed <see cref="AuditEvent.EventId"/> at that instant, or null.</param>
@@ -47,6 +47,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
    private const string ReconnectTimerKey = "grpc-reconnect";
    private const string StabilityTimerKey = "grpc-stability";
    private const string SnapshotTimerKey = "debug-snapshot-deadline";
    private const string ConsumerLivenessTimerKey = "debug-consumer-liveness";
    /// <summary>Delay between gRPC reconnection attempts.</summary>
    internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);
@@ -69,6 +70,33 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
    /// </summary>
    internal static TimeSpan StabilityWindow { get; set; } = TimeSpan.FromSeconds(60);
    /// <summary>
    /// Orphan window: how long the session may go without ANY sign of life from its CONSUMER
    /// before it self-terminates. Renewed by <see cref="DebugStreamConsumerAlive"/>, which
    /// <c>DebugStreamService</c> Tells on a timer for every session still attached to a
    /// consumer (Blazor debug view or the SignalR hub) — so it measures the consumer, never
    /// the stream. Settable for tests.
    /// </summary>
    internal static TimeSpan ConsumerIdleTimeout { get; set; } = TimeSpan.FromMinutes(5);
    /// <summary>
    /// How often the actor checks the consumer-last-seen stamp against
    /// <see cref="ConsumerIdleTimeout"/>. A self-tick rather than <c>SetReceiveTimeout</c>:
    /// the receive timeout measures the MAILBOX, which conflates site chatter with consumer
    /// liveness — and once stream events were correctly excluded from it (via
    /// <see cref="LiveDebugStreamEvent"/>) nothing recurring reset it at all, so every healthy
    /// session self-terminated one window after its snapshot with a false "Site disconnected".
    /// Settable for tests.
    /// </summary>
    internal static TimeSpan ConsumerLivenessCheckInterval { get; set; } = TimeSpan.FromSeconds(30);
    /// <summary>
    /// When the consumer was last known to be attached (UTC). Seeded in <see cref="PreStart"/>
    /// so a session gets a full window to receive its first keepalive, then refreshed by every
    /// <see cref="DebugStreamConsumerAlive"/>. Actor-thread only.
    /// </summary>
    private DateTime _consumerLastSeenUtc = DateTime.UtcNow;
    private int _retryCount;
    private bool _useNodeA = true;
    private bool _stopped;
@@ -188,7 +216,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
                // non-deployed instance — cancel it (and any buffered gap events are
                // discarded with the actor). No pass-through.
                // _stopped is set AFTER CleanupGrpc() to match the ordering in the
-                // DebugStreamTerminated and ReceiveTimeout handlers (cosmetic consistency).
+                // DebugStreamTerminated and consumer-liveness handlers (cosmetic consistency).
                CleanupGrpc();
                _stopped = true;
                _preSnapshotBuffer.Clear();
@@ -217,8 +245,8 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
        // Hard snapshot deadline (WP2.3). Nothing else ends a session stuck in the
        // buffering phase: the site's reply was lost, so no gRPC error fires, the stream
-        // keeps delivering events, and (with the wrapper above) they no longer even reset
+        // keeps delivering events, and stream traffic does not renew the orphan net (which
-        // the orphan timeout. Fail the session so the consumer is told and can reopen.
+        // measures the consumer). Fail the session so the consumer is told and can reopen.
        Receive<DebugSnapshotDeadline>(_ =>
        {
            if (_stopped || _snapshotDelivered) return;
@@ -234,21 +262,19 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
            Context.Stop(Self);
        });
-        // Domain events arriving via Self.Tell from the gRPC callback, wrapped so they do
+        // Domain events arriving via Self.Tell from the gRPC callback. Stream traffic never
-        // NOT influence the receive timeout (WP2.3): the orphan safety net exists to end a
+        // proves the CONSUMER is still there, so it deliberately does not touch the orphan
-        // session whose CONSUMER is gone, and a busy site's event flood used to keep that
+        // net (which now measures the consumer keepalive, not the mailbox). Receiving an
-        // net permanently reset — an abandoned session on a chatty instance never timed out.
+        // event must not reset _retryCount either: a flapping stream that delivers a single
-        // Receiving an event must not reset _retryCount either: a flapping stream that
+        // event between failures would otherwise never trip MaxRetries. The retry budget is
-        // delivers a single event between failures would otherwise never trip MaxRetries.
+        // recovered only by GrpcStreamStable (a stream that has stayed up for
-        // The retry budget is recovered only by GrpcStreamStable (a stream that has stayed
+        // StabilityWindow). Before the snapshot has been delivered, BUFFER (in arrival order)
-        // up for StabilityWindow). Before the snapshot has been delivered, BUFFER (in arrival
+        // rather than deliver — these may be gap-window events; after the snapshot has been
-        // order) rather than deliver — these may be gap-window events; after the snapshot has
+        // flushed, pass through directly (phase-dependent behavior).
        // been flushed, pass through directly (phase-dependent behavior).
        Receive<LiveDebugStreamEvent>(wrapped => HandleStreamEvent(wrapped.Event));
        // Unwrapped forms are still accepted (a direct Tell from a test or a future
-        // in-process producer); those DO influence the receive timeout, which is correct —
+        // in-process producer) and take the identical path.
        // they are not the high-volume stream path.
        Receive<AttributeValueChanged>(changed => HandleStreamEvent(changed));
        Receive<AlarmStateChanged>(changed => HandleStreamEvent(changed));
@@ -315,11 +341,32 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
            Context.Stop(Self);
        });
-        // Orphan safety net — if nobody stops us within 5 minutes, self-terminate
+        // Consumer keepalive: DebugStreamService Tells this on a timer for every session it
-        Context.SetReceiveTimeout(TimeSpan.FromMinutes(5));
+        // still holds (i.e. still attached to a Blazor debug view / SignalR connection). It
-        Receive<ReceiveTimeout>(_ =>
+        // is the ONLY thing that renews the orphan window — deliberately, so neither a chatty
        // site nor a silent one can influence it.
        Receive<DebugStreamConsumerAlive>(_ =>
        {
-            _log.Warning("Debug stream for {0} timed out (orphaned session), stopping", _instanceUniqueName);
+            if (_stopped) return;
            _consumerLastSeenUtc = DateTime.UtcNow;
        });
        // Orphan safety net, CONSUMER-measured (WP2.3 follow-up). A periodic self-tick
        // compares the consumer-last-seen stamp against ConsumerIdleTimeout; the previous
        // SetReceiveTimeout(5 min) measured the mailbox instead, and once stream events were
        // (correctly) marked INotInfluenceReceiveTimeout nothing recurring reset it — the
        // snapshot arrives once and GrpcStreamStable once, so EVERY healthy session died at
        // ~6 minutes and the consumer was told "Site disconnected".
        Receive<ConsumerLivenessTick>(_ =>
        {
            if (_stopped) return;
            var idle = DateTime.UtcNow - _consumerLastSeenUtc;
            if (idle < ConsumerIdleTimeout) return;
            _log.Warning(
                "Debug stream for {0} has had no consumer activity for {1:F0}s (orphaned session), stopping",
                _instanceUniqueName, idle.TotalSeconds);
            Timers.Cancel(ConsumerLivenessTimerKey);
            CleanupGrpc();
            SendUnsubscribe();
            _stopped = true;
@@ -507,6 +554,15 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
        // Arm the hard snapshot deadline alongside the request.
        if (SnapshotTimeout > TimeSpan.Zero)
            Timers.StartSingleTimer(SnapshotTimerKey, new DebugSnapshotDeadline(), SnapshotTimeout);
        // Arm the consumer-liveness net. The stamp starts now, so the session always gets a
        // full ConsumerIdleTimeout to see its first keepalive from DebugStreamService.
        _consumerLastSeenUtc = DateTime.UtcNow;
        if (ConsumerIdleTimeout > TimeSpan.Zero && ConsumerLivenessCheckInterval > TimeSpan.Zero)
        {
            Timers.StartPeriodicTimer(
                ConsumerLivenessTimerKey, new ConsumerLivenessTick(), ConsumerLivenessCheckInterval);
        }
    }
    /// <inheritdoc />
@@ -546,7 +602,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
            await client.SubscribeAsync(
                _correlationId,
                _instanceUniqueName,
-                // Wrapped: stream traffic must not reset the orphan receive timeout.
+                // Wrapped so the stream path is explicit (it never renews the orphan net).
                evt => self.Tell(new LiveDebugStreamEvent(evt)),
                ex => self.Tell(new GrpcStreamError(ex, generation)),
                () => self.Tell(new GrpcStreamCompleted(generation)),
@@ -660,11 +716,26 @@ public record StopDebugStream;
 /// <summary>
 /// Envelope for a live gRPC stream event (<c>AttributeValueChanged</c>/
-/// <c>AlarmStateChanged</c>). Implements <see cref="INotInfluenceReceiveTimeout"/> so a busy
+/// <c>AlarmStateChanged</c>). Kept as a distinct envelope so the high-volume stream path is
-/// site's event flood cannot keep resetting the orphan-session receive timeout — the timeout
+/// explicit at the call site; the orphan net no longer keys off the mailbox at all (it
-/// measures consumer/session liveness, not site chatter (WP2.3).
+/// measures the consumer keepalive), so a busy site's event flood can neither hold a dead
 /// session open nor — as briefly happened — be the only thing keeping a healthy one alive.
 /// </summary>
-internal record LiveDebugStreamEvent(object Event) : INotInfluenceReceiveTimeout;
+internal record LiveDebugStreamEvent(object Event);
 /// <summary>
 /// Consumer keepalive: <c>DebugStreamService</c> Tells one of these to every bridge actor
 /// whose session is still attached to a consumer, on
 /// <see cref="DebugStreamBridgeActor.ConsumerLivenessCheckInterval"/>-scale cadence. Renewing
 /// the consumer-last-seen stamp is its ONLY effect.
 /// </summary>
 public record DebugStreamConsumerAlive;
 /// <summary>
 /// Internal self-tick that checks the consumer-last-seen stamp against
 /// <see cref="DebugStreamBridgeActor.ConsumerIdleTimeout"/>.
 /// </summary>
 internal record ConsumerLivenessTick;
 /// <summary>
 /// Internal message: the hard deadline for the initial <c>DebugViewSnapshot</c> expired.
@@ -144,15 +144,38 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
    private bool _seedOnConnect;
    /// <summary>
-    /// Set whenever a fan-out finishes (success or failure); consumed by the next reconcile
+    /// Set when a CONNECT- OR FAILOVER-DRIVEN fan-out finishes (the initial seed, a re-seed
-    /// tick, which skips its own fan-out when it finds the flag set. That makes the
+    /// consumed by a successful (re)connect, or a re-seed queued behind one); consumed by the
-    /// connect-driven seed and the periodic backstop mutually exclusive — the pair used to
+    /// next reconcile tick, which skips its own fan-out when it finds the flag set. That makes
    /// the connect-driven seed and the periodic backstop mutually exclusive — the pair used to
    /// run BOTH, which is the "full unconditional snapshot every 60s" waste — while bounding
    /// staleness at two intervals (a tick can be skipped at most once in a row).
    /// <para>
    /// A TICK-driven fan-out deliberately does NOT set it. It used to: the tick's own fan-out
    /// completion armed the skip, so steady state ran fan-out → skip → fan-out → skip, i.e.
    /// one reconcile per TWO intervals — halving the not-reporting refresh rate and the alarm
    /// reconcile backstop for no reason (nothing was duplicated to suppress).
    /// </para>
    /// Actor-thread only.
    /// </summary>
    private bool _fanoutSinceLastTick;
    /// <summary>
    /// Whether the in-flight fan-out is connect/failover-driven and therefore makes the next
    /// reconcile tick redundant (see <see cref="_fanoutSinceLastTick"/>). Stable for the
    /// fan-out's lifetime: <see cref="StartFanout"/> never overwrites it while one is in
    /// flight (a colliding request queues instead). Actor-thread only.
    /// </summary>
    private bool _fanoutSuppressesNextTick;
    /// <summary>
    /// Same flag for a fan-out that was queued behind an in-flight one
    /// (<see cref="_reseedQueued"/>) — OR-ed across colliding requests so a queued
    /// connect/failover re-seed keeps its suppression even if a tick collided too.
    /// Actor-thread only.
    /// </summary>
    private bool _queuedFanoutSuppressesNextTick;
    /// <summary>
    /// True between issuing a stream open and its first outcome (connected / error /
    /// completed), so a reconcile tick never stacks a second open on an in-flight one.
@@ -269,7 +292,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        Receive<RetrySeed>(_ =>
        {
            if (_stopped) return;
-            StartFanout(isInitial: false);
+            // A backoff retry of a failed fan-out is not a connect/failover seed: it must
            // not stand the next reconcile tick down.
            StartFanout(isInitial: false, suppressesNextTick: false);
        });
        // The site-wide stream is confirmed established (response headers received). This —
@@ -356,7 +381,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        // Kick the initial seed fan-out. The connect callback for this first stream must
        // NOT seed again on top of it, so the flag starts cleared.
        _seedOnConnect = false;
-        StartFanout(isInitial: true);
+        StartFanout(isInitial: true, suppressesNextTick: true);
        // Periodic reconcile backstop. Single-shot and re-armed with fresh jitter each
        // tick (a periodic timer would lock every site to the same phase forever).
@@ -398,9 +423,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
    /// <summary>
    /// Periodic reconcile backstop: re-run the snapshot fan-out (corrects instance-set drift
-    /// + any missed delta) UNLESS one already completed inside this window — a reconnect-driven
+    /// + any missed delta) UNLESS a CONNECT- OR FAILOVER-DRIVEN seed already completed inside
-    /// seed makes the tick redundant, and running both was the "full unconditional snapshot
+    /// this window — that seed makes the tick redundant, and running both was the "full
-    /// every 60s" waste (WP2.3). If the live stream was previously given up, self-heal it by
+    /// unconditional snapshot every 60s" waste (WP2.3). A tick's OWN fan-out never arms that
    /// skip: it used to, which made steady state alternate fan-out/skip and halved the
    /// effective reconcile rate. If the live stream was previously given up, self-heal it by
    /// resetting the retry budget and reopening, so a sustained outage never permanently kills
    /// the live feed.
    /// </summary>
@@ -441,7 +468,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        }
        if (!reopening)
-            StartFanout(isInitial: false);
+        {
            // Tick-driven: deliberately does NOT set the skip flag, so the NEXT tick fans
            // out too. Setting it here was the "one reconcile per two intervals" bug.
            StartFanout(isInitial: false, suppressesNextTick: false);
        }
    }
    // ── Seed / reconcile fan-out ────────────────────────────────────────────────
@@ -451,7 +482,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
    /// <c>Self.Tell</c>. While in flight, live deltas buffer. A reconcile that arrives
    /// while a fan-out is already running is skipped (no stacking).
    /// </summary>
-    private void StartFanout(bool isInitial)
+    private void StartFanout(bool isInitial, bool suppressesNextTick)
    {
        if (_stopped) return;
        if (_fanoutInFlight)
@@ -460,11 +491,16 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
            // one — its snapshot read-time predates the stream death, so skipping it would
            // serve stale up to the next 60s reconcile (N7.1). An initial-seed collision
            // never queues (there is only ever one).
-            if (!isInitial) _reseedQueued = true;
+            if (!isInitial)
            {
                _reseedQueued = true;
                _queuedFanoutSuppressesNextTick |= suppressesNextTick;
            }
            return;
        }
        _fanoutInFlight = true;
        _fanoutSuppressesNextTick = suppressesNextTick;
        var self = Self;
        var ct = _lifetimeCts?.Token ?? CancellationToken.None;
@@ -514,7 +550,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        var flushChanged = FlushBuffer();
        _fanoutInFlight = false;
-        _fanoutSinceLastTick = true;
+        // Only a connect/failover-driven seed makes the next tick redundant — a tick's own
        // fan-out must never suppress the tick after it (that halved the reconcile rate).
        if (_fanoutSuppressesNextTick) _fanoutSinceLastTick = true;
        _consecutiveSeedFailures = 0;
        Timers.Cancel(SeedRetryTimerKey);
        var firstSeed = !_seeded;
@@ -531,11 +569,21 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
            Publish();
        // A failover re-seed requested while this fan-out was in flight runs now (N7.1).
-        if (_reseedQueued)
+        RunQueuedFanoutIfAny();
-        {
+    }
-            _reseedQueued = false;
+
-            StartFanout(isInitial: false);
+    /// <summary>
-        }
+    /// Runs a fan-out that collided with an in-flight one, carrying the suppression flag it
    /// was queued with so a queued connect/failover re-seed still stands the next tick down
    /// (and a queued TICK fan-out still does not).
    /// </summary>
    private void RunQueuedFanoutIfAny()
    {
        if (!_reseedQueued) return;
        _reseedQueued = false;
        var suppresses = _queuedFanoutSuppressesNextTick;
        _queuedFanoutSuppressesNextTick = false;
        StartFanout(isInitial: false, suppressesNextTick: suppresses);
    }
    /// <summary>
@@ -570,7 +618,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        // Don't lose deltas captured during the failed window — apply them pass-through
        // into the (possibly stale/empty) cache. The next reconcile re-seeds authoritatively.
        _fanoutInFlight = false;
-        _fanoutSinceLastTick = true;
+        // Same rule as the success path: only a connect/failover-driven fan-out stands the
        // next reconcile tick down.
        if (_fanoutSuppressesNextTick) _fanoutSinceLastTick = true;
        var flushChanged = FlushBuffer(dedupAgainstSeed: false);
        // Only publish if we already had a seed (so IsLive doesn't flip true on a
@@ -585,8 +635,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        // A failover re-seed requested while this fan-out was in flight runs now (N7.1).
        if (_reseedQueued)
        {
-            _reseedQueued = false;
+            RunQueuedFanoutIfAny();
            StartFanout(isInitial: false);
            return;
        }
@@ -794,7 +843,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        {
            _seedOnConnect = false;
            _log.Info("Site-alarm gRPC stream for {0} (re)connected; running one re-seed", _siteIdentifier);
-            StartFanout(isInitial: false);
+            StartFanout(isInitial: false, suppressesNextTick: true);
        }
        // Liveness recovered — republish so viewers stop falling back to polling even if the
@@ -13,14 +13,34 @@ namespace ZB.MOM.WW.ScadaBridge.Communication;
 /// Manages debug stream sessions by creating DebugStreamBridgeActors that persist
 /// as subscribers on the site side. Both the Blazor debug view and the SignalR hub
 /// use this service to start/stop streams.
 /// <para>
 /// <b>Consumer keepalive.</b> This service is the session registry, and holding a session
 /// here IS what "a consumer is attached" means: the Blazor debug view and the SignalR hub
 /// both release their session (<see cref="StopStream"/>) on dispose/disconnect. A single
 /// shared timer therefore Tells <see cref="DebugStreamConsumerAlive"/> to every registered
 /// bridge actor, which is the only thing that renews each actor's orphan window. A bridge
 /// actor that outlives its registration — the leak the orphan net exists for — stops
 /// receiving keepalives and self-terminates. The keepalive is independent of stream traffic
 /// by design: neither a chatty nor a silent site can influence the orphan decision.
 /// </para>
 /// </summary>
-public class DebugStreamService
+public class DebugStreamService : IDisposable
 {
    /// <summary>
    /// Cadence of the shared consumer keepalive. Comfortably shorter than
    /// <see cref="DebugStreamBridgeActor.ConsumerIdleTimeout"/> so a few missed ticks (GC
    /// pause, thread-pool starvation) can never orphan a live session. Settable for tests.
    /// </summary>
    internal static TimeSpan KeepaliveInterval { get; set; } = TimeSpan.FromSeconds(30);
    private readonly CommunicationService _communicationService;
    private readonly IServiceProvider _serviceProvider;
    private readonly SiteStreamGrpcClientFactory _grpcClientFactory;
    private readonly ILogger<DebugStreamService> _logger;
    private readonly ConcurrentDictionary<string, IActorRef> _sessions = new();
    private readonly object _keepaliveLock = new();
    private Timer? _keepaliveTimer;
    private bool _disposed;
    private ActorSystem? _actorSystem;
    /// <summary>
@@ -135,6 +155,7 @@ public class DebugStreamService
        var bridgeActor = system.ActorOf(props, $"debug-stream-{sessionId}");
        _sessions[sessionId] = bridgeActor;
        EnsureKeepaliveTimer();
        // Wait for the initial snapshot (with timeout)
        using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
@@ -184,6 +205,73 @@ public class DebugStreamService
            _logger.LogInformation("Debug stream {SessionId} stopped", sessionId);
        }
    }
    /// <summary>
    /// Sends one keepalive round to every attached session. Exposed (internal) so tests can
    /// drive the keepalive deterministically instead of waiting on the timer.
    /// </summary>
    internal void SendConsumerKeepalives()
    {
        foreach (var session in _sessions)
        {
            session.Value.Tell(new DebugStreamConsumerAlive());
        }
    }
    /// <summary>
    /// Starts the shared keepalive timer on first use. One timer for the whole service (not
    /// one per session): the payload is a Tell per attached session, so a single periodic
    /// callback is the cheapest shape that covers every consumer surface.
    /// </summary>
    private void EnsureKeepaliveTimer()
    {
        if (_keepaliveTimer is not null) return;
        lock (_keepaliveLock)
        {
            if (_keepaliveTimer is not null || _disposed) return;
            _keepaliveTimer = new Timer(
                _ =>
                {
                    try
                    {
                        SendConsumerKeepalives();
                    }
                    catch (Exception ex)
                    {
                        // A keepalive round must never take the timer down — a thrown Tell
                        // would silently stop renewing EVERY session's orphan window.
                        _logger.LogWarning(ex, "Debug stream consumer keepalive round failed");
                    }
                },
                null,
                KeepaliveInterval,
                KeepaliveInterval);
        }
    }
    /// <summary>Stops the keepalive timer.</summary>
    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }
    /// <summary>Stops the keepalive timer.</summary>
    /// <param name="disposing">True when called from <see cref="Dispose()"/>.</param>
    protected virtual void Dispose(bool disposing)
    {
        if (_disposed) return;
        lock (_keepaliveLock)
        {
            if (_disposed) return;
            _disposed = true;
            if (disposing)
            {
                _keepaliveTimer?.Dispose();
                _keepaliveTimer = null;
            }
        }
    }
 }
 public record DebugStreamSession(string SessionId, DebugViewSnapshot InitialSnapshot);
@@ -246,11 +246,12 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
    /// </param>
    /// <param name="ct">Cancellation token to stop the subscription.</param>
    /// <param name="onConnected">
-    /// Optional callback invoked once when the site has ACCEPTED the subscription (response
+    /// Optional callback invoked once when the site has ACCEPTED the subscription — response
-    /// headers received — the site writes them as soon as its relay actor is subscribed, so
+    /// headers received (the site writes them as soon as its relay actor is subscribed, so no
-    /// no event can be missed after this point). The per-site aggregator uses it to run
+    /// event can be missed after this point), or the first event received if the peer defers
-    /// exactly one re-seed per successful (re)connect instead of one per reconnect attempt.
+    /// its headers. A header timeout is NOT treated as connected. The per-site aggregator uses
-    /// Never invoked more than once per call, and never after <paramref name="onError"/>.
+    /// this to run exactly one re-seed per successful (re)connect instead of one per reconnect
    /// attempt. Never invoked more than once per call, and never after <paramref name="onError"/>.
    /// </param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public virtual async Task SubscribeSiteAsync(
@@ -311,11 +312,11 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
    /// <param name="onError">Invoked once if the stream faulted.</param>
    /// <param name="onCompleted">Invoked once if the server ended the stream with OK.</param>
    /// <param name="onConnected">
-    /// Optional; invoked once when the server's response headers arrive — i.e. the site has
+    /// Optional; invoked AT MOST ONCE when the site has demonstrably accepted the
-    /// accepted the subscription and its relay actor is attached. Bounded by
+    /// subscription — either the server's response headers arrived (bounded by
-    /// <see cref="ConnectedHeaderTimeout"/> so a peer that defers headers (a pre-WP2.3 site,
+    /// <see cref="ConnectedHeaderTimeout"/>) or, for a peer that defers headers until its
-    /// which only flushes them with its first event) still reports connected instead of
+    /// first message, the first event was received. A header timeout alone is never
-    /// leaving the caller waiting for a signal that may never come on a quiet site.
+    /// reported as connected: that is also exactly what an unreachable site looks like.
    /// </param>
    /// <returns>A task that completes when the stream has ended and its outcome been reported.</returns>
    internal async Task ConsumeStreamAsync(
@@ -328,27 +329,49 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
        Action? onConnected = null)
    {
        var completedGracefully = false;
        var connectedReported = false;
        // Fires onConnected AT MOST ONCE, from whichever proof of a live peer arrives
        // first: the response headers, or (for a peer that defers headers until its first
        // message) the first event. A header TIMEOUT is deliberately NOT such a proof —
        // see AwaitHeadersAsync.
        void ReportConnected()
        {
            if (connectedReported || onConnected is null) return;
            connectedReported = true;
            onConnected();
        }
        try
        {
            using (var call = openCall())
            {
-                if (onConnected is not null)
+                if (onConnected is not null && await AwaitHeadersAsync(call, cts.Token).ConfigureAwait(false))
                {
-                    await AwaitHeadersAsync(call, cts.Token).ConfigureAwait(false);
+                    ReportConnected();
                    onConnected();
                }
                await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token))
                {
                    // Fallback connected signal: an event can only come from a peer that
                    // accepted the subscription, so it proves what the headers would have.
                    // Raised BEFORE the event is delivered so the consumer sees
                    // connected-then-event ordering.
                    ReportConnected();
                    onEvent(evt);
                }
            }
            completedGracefully = true;
        }
-        catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
+        catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled && cts.IsCancellationRequested)
        {
-            // Normal cancellation — not an error
+            // OUR OWN cancellation (Unsubscribe / reconnect / channel teardown we asked
            // for) — not an error. The IsCancellationRequested guard matters: a Cancelled
            // status can also originate at the PEER or from a channel disposed underneath
            // us, and swallowing THAT fired none of onError/onCompleted/onConnected, so
            // the consuming actor kept a dead stream marked live forever. Foreign
            // Cancelled now falls through to the onError path below.
        }
        catch (OperationCanceledException) when (cts.IsCancellationRequested)
        {
@@ -372,26 +395,37 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
    }
    /// <summary>
-    /// How long to wait for response headers before treating the stream as connected anyway.
+    /// How long to wait for response headers before giving up on them as the connected
-    /// A peer that only flushes headers with its first message would otherwise hold the
+    /// signal and falling back to the first received event. A peer that only flushes headers
-    /// connected signal — and with it the aggregator's re-seed — for as long as the site
+    /// with its first message would otherwise hold the connected signal — and with it the
-    /// happens to be quiet.
+    /// aggregator's re-seed — for as long as the site happens to be quiet.
    /// </summary>
    internal static TimeSpan ConnectedHeaderTimeout { get; set; } = TimeSpan.FromSeconds(10);
    /// <summary>
    /// Awaits the call's response headers, bounded by <see cref="ConnectedHeaderTimeout"/>.
-    /// A fault propagates (the caller reports it through <c>onError</c> like any other stream
+    /// Returns <see langword="true"/> when the headers arrived (the site accepted the
-    /// fault); a timeout returns normally. On timeout the abandoned headers task is observed
+    /// subscription) and <see langword="false"/> on timeout. A fault propagates (the caller
-    /// so a later fault on it can never surface as an unobserved task exception.
+    /// reports it through <c>onError</c> like any other stream fault); on timeout the
    /// abandoned headers task is observed so a later fault on it can never surface as an
    /// unobserved task exception.
    /// <para>
    /// A timeout must NOT be reported as connected: an unreachable/wedged site produces
    /// exactly that shape, and calling <c>onConnected</c> for it made the aggregator clear
    /// <c>_streamDown</c>, consume its pending re-seed and fan a full snapshot out at a site
    /// that never answered. The caller instead treats the FIRST RECEIVED EVENT as the
    /// fallback connected signal — real proof of a live peer, and the quiet-site case the
    /// timeout was added for is covered by the reconcile backstop.
    /// </para>
    /// </summary>
-    private static async Task AwaitHeadersAsync(
+    private static async Task<bool> AwaitHeadersAsync(
        AsyncServerStreamingCall<SiteStreamEvent> call, CancellationToken ct)
    {
        var headers = call.ResponseHeadersAsync;
        try
        {
            await headers.WaitAsync(ConnectedHeaderTimeout, ct).ConfigureAwait(false);
            return true;
        }
        catch (TimeoutException)
        {
@@ -400,6 +434,7 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
                CancellationToken.None,
                TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
                TaskScheduler.Default);
            return false;
        }
    }
@@ -1,4 +1,5 @@
 using System.Diagnostics;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 using Microsoft.Extensions.Options;
 using NSubstitute;
@@ -115,4 +116,67 @@ public class SiteAuditBacklogReporterCadenceTests
        Assert.Equal(TimeSpan.FromSeconds(3), reporter.RefreshInterval);
    }
    // ----- Stale-Pending signal (review F5) ----- //
    [Fact]
    public void StalePendingBacklog_IsWarned_ThenRateLimited()
    {
        // Pending rows are exempt from the retention purge by design, so a standing Pending
        // backlog is the one site-store condition that never self-heals on age — it clears
        // only when central acknowledges the rows. It is worth a log line, not just a number
        // on the health report, and the warning must not spam every 30 s poll.
        var logger = new CapturingLogger();
        var reporter = new SiteAuditBacklogReporter(
            Substitute.For<ISiteAuditQueue>(),
            Substitute.For<ISiteHealthCollector>(),
            logger,
            TimeSpan.FromHours(1),
            null);
        var stale = DateTime.UtcNow - SiteAuditBacklogReporter.StalePendingThreshold - TimeSpan.FromHours(1);
        reporter.WarnIfPendingIsStale(stale, pendingCount: 4321);
        reporter.WarnIfPendingIsStale(stale, pendingCount: 4321); // same poll cycle-ish
        var warning = Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning);
        Assert.Contains("4321", warning.Message);
        Assert.Contains("pending", warning.Message, StringComparison.OrdinalIgnoreCase);
    }
    [Fact]
    public void FreshOrEmptyPendingBacklog_IsNotWarned()
    {
        var logger = new CapturingLogger();
        var reporter = new SiteAuditBacklogReporter(
            Substitute.For<ISiteAuditQueue>(),
            Substitute.For<ISiteHealthCollector>(),
            logger,
            TimeSpan.FromHours(1),
            null);
        reporter.WarnIfPendingIsStale(null, pendingCount: 0);                    // nothing pending
        reporter.WarnIfPendingIsStale(DateTime.UtcNow.AddMinutes(-5), 12);       // a normal drain lag
        Assert.DoesNotContain(logger.Entries, e => e.Level == LogLevel.Warning);
    }
    /// <summary>Captures log entries so the stale-pending signal can be asserted.</summary>
    private sealed class CapturingLogger : ILogger<SiteAuditBacklogReporter>
    {
        public List<(LogLevel Level, Exception? Exception, string Message)> Entries { get; } = new();
        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
        public bool IsEnabled(LogLevel logLevel) => true;
        public void Log<TState>(
            LogLevel logLevel,
            EventId eventId,
            TState state,
            Exception? exception,
            Func<TState, Exception?, string> formatter)
        {
            Entries.Add((logLevel, exception, formatter(state, exception)));
        }
    }
 }
@@ -614,6 +614,10 @@ public class SqliteAuditWriterWriteTests
        await writer.WriteAsync(older);
        await writer.WriteAsync(boundary);
        // Serve them first — retirement is bounded by what has actually been served, which
        // is exactly the order the pull handler runs in (read, then next pull's cursor).
        await writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 100);
        var flipped = await writer.MarkReconciledUpToAsync(instant, afterId: null);
        Assert.Equal(1, flipped);
@@ -648,6 +652,99 @@ public class SqliteAuditWriterWriteTests
            second.Select(r => r.EventId).ToHashSet());
    }
    [Fact]
    public async Task MarkReconciledUpToAsync_LateStampedInsertBelowTheCursor_IsNeverRetired()
    {
        // THE data-loss case. OccurredAtUtc is caller-stamped, so a row can be INSERTED
        // after a batch was served yet carry a timestamp BELOW central's (by then advanced)
        // cursor — a script that back-dates, a clock nudge, a queued write flushed late.
        // The blanket cursor UPDATE retired exactly those rows: never served, never servable
        // again (the keyset read has moved past them) and, being Reconciled, purged on age.
        // The insertion-order (rowid) bound makes them unreachable by the flip.
        var (writer, dataSource) = CreateWriter(
            nameof(MarkReconciledUpToAsync_LateStampedInsertBelowTheCursor_IsNeverRetired));
        await using var _w = writer;
        var t0 = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
        var served1 = NewEvent(occurredAtUtc: t0);
        var served2 = NewEvent(occurredAtUtc: t0.AddSeconds(20));
        await writer.WriteAsync(served1);
        await writer.WriteAsync(served2);
        // Central pulls both and advances its cursor to the newest row.
        var page = await writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 100);
        Assert.Equal(2, page.Count);
        var cursorTime = t0.AddSeconds(20);
        var cursorId = served2.EventId.ToString();
        // A row lands NOW carrying a timestamp between the two served rows.
        var lateStamped = NewEvent(occurredAtUtc: t0.AddSeconds(10));
        await writer.WriteAsync(lateStamped);
        var flipped = await writer.MarkReconciledUpToAsync(cursorTime, cursorId);
        // The two genuinely served rows retire; the late-stamped one does not.
        Assert.Equal(2, flipped);
        Assert.Equal(AuditForwardState.Reconciled.ToString(), ReadForwardState(dataSource, served1.EventId));
        Assert.Equal(AuditForwardState.Reconciled.ToString(), ReadForwardState(dataSource, served2.EventId));
        Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, lateStamped.EventId));
        // Still recoverable: a central that restarts (cursor resets to MinValue) re-serves it,
        // and — being Pending — the retention purge can never drop it in the meantime.
        var reread = await writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 100);
        Assert.Equal(lateStamped.EventId, Assert.Single(reread).EventId);
    }
    [Fact]
    public async Task MarkReconciledUpToAsync_ForwardedRowsRetire_EvenWithoutHavingBeenPulled()
    {
        // The bound applies to PENDING rows (nothing proves central saw them but the pull).
        // A FORWARDED row was ACKED by central through the telemetry push path, so the cursor
        // may retire it whether or not this node has ever served it in a pull — otherwise a
        // site node that never serves a pull would accumulate acked rows forever.
        var (writer, dataSource) = CreateWriter(
            nameof(MarkReconciledUpToAsync_ForwardedRowsRetire_EvenWithoutHavingBeenPulled));
        await using var _w = writer;
        var t0 = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
        var pushed = NewEvent(occurredAtUtc: t0);
        var neverShipped = NewEvent(occurredAtUtc: t0.AddSeconds(1));
        await writer.WriteAsync(pushed);
        await writer.WriteAsync(neverShipped);
        // Central acked the first row over the telemetry drain — no pull involved.
        await writer.MarkForwardedAsync(new[] { pushed.EventId });
        var flipped = await writer.MarkReconciledUpToAsync(t0.AddSeconds(30), afterId: null);
        Assert.Equal(1, flipped);
        Assert.Equal(AuditForwardState.Reconciled.ToString(), ReadForwardState(dataSource, pushed.EventId));
        Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, neverShipped.EventId));
    }
    [Fact]
    public async Task MarkReconciledUpToAsync_BeforeAnyPull_RetiresNothingPending()
    {
        // Bound state is per-process: after a site-node restart nothing has been served yet,
        // so an incoming cursor retires no Pending row. Conservative in the safe direction —
        // the rows stay servable and the next pull re-establishes the bound.
        var (writer, dataSource) = CreateWriter(nameof(MarkReconciledUpToAsync_BeforeAnyPull_RetiresNothingPending));
        await using var _w = writer;
        var t0 = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
        var evt = NewEvent(occurredAtUtc: t0);
        await writer.WriteAsync(evt);
        var flipped = await writer.MarkReconciledUpToAsync(t0.AddHours(1), afterId: null);
        Assert.Equal(0, flipped);
        Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, evt.EventId));
        // …and the very next pull cycle retires it normally.
        await writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 100);
        Assert.Equal(1, await writer.MarkReconciledUpToAsync(t0.AddHours(1), afterId: null));
    }
    [Fact]
    public async Task ReadPendingSinceAsync_InvalidBatchSize_Throws()
    {
@@ -16,6 +16,11 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
 /// <summary>
 /// Tests for DebugStreamService session lifecycle.
 /// </summary>
 /// <remarks>
 /// Shares the <c>DebugStreamStatics</c> xUnit collection with <c>DebugStreamBridgeActorTests</c>
 /// so the two classes never race on the actor's static test seams (see that class).
 /// </remarks>
 [Collection("DebugStreamStatics")]
 public class DebugStreamServiceTests : TestKit
 {
    [Fact]
@@ -74,4 +79,89 @@ public class DebugStreamServiceTests : TestKit
        Assert.Contains("Site1.Pump01", ex.Message);
        Assert.NotNull(ex.InnerException);
    }
    [Fact]
    public async Task AttachedSession_IsKeptAliveByTheServiceKeepalive_AndDiesOnceDetached()
    {
        // The consumer half of the orphan net: holding a session in DebugStreamService IS
        // "a consumer is attached", and the service's shared timer is what renews the bridge
        // actor's window. Without this wiring every healthy session self-terminated one
        // window after its snapshot (the old mailbox-based ReceiveTimeout had nothing
        // recurring to reset it) and the consumer was told "Site disconnected".
        var previousKeepalive = DebugStreamService.KeepaliveInterval;
        var previousIdle = DebugStreamBridgeActor.ConsumerIdleTimeout;
        var previousCheck = DebugStreamBridgeActor.ConsumerLivenessCheckInterval;
        DebugStreamService.KeepaliveInterval = TimeSpan.FromMilliseconds(50);
        DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMilliseconds(400);
        DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromMilliseconds(50);
        try
        {
            var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 };
            var site = new Site("Site One", "site-1")
            {
                Id = 3,
                GrpcNodeAAddress = "http://localhost:5100",
                GrpcNodeBAddress = "http://localhost:5200"
            };
            var instanceRepo = Substitute.For<ITemplateEngineRepository>();
            instanceRepo.GetInstanceByIdAsync(7, Arg.Any<CancellationToken>()).Returns(instance);
            var siteRepo = Substitute.For<ISiteRepository>();
            siteRepo.GetSiteByIdAsync(3, Arg.Any<CancellationToken>()).Returns(site);
            var services = new ServiceCollection();
            services.AddScoped(_ => instanceRepo);
            services.AddScoped(_ => siteRepo);
            using var provider = services.BuildServiceProvider();
            var commProbe = CreateTestProbe();
            var commService = new CommunicationService(
                Options.Create(new CommunicationOptions()),
                NullLogger<CommunicationService>.Instance);
            commService.SetCommunicationActor(commProbe.Ref);
            // Mock gRPC factory: the real one would dial localhost:5100, fail, and trip the
            // bridge actor's retry budget — a termination unrelated to the orphan net under
            // test. The mock keeps the stream "up" so the only thing that can end this
            // session is the consumer-liveness decision.
            using var grpcFactory = new Grpc.MockSiteStreamGrpcClientFactory(
                new Grpc.MockSiteStreamGrpcClient());
            using var service = new DebugStreamService(
                commService, provider, grpcFactory, NullLogger<DebugStreamService>.Instance);
            service.SetActorSystem(Sys);
            var startTask = service.StartStreamAsync(
                instanceId: 7, onEvent: _ => { }, onTerminated: () => { });
            commProbe.ExpectMsg<SiteEnvelope>(TimeSpan.FromSeconds(5));
            var bridgeActor = commProbe.LastSender;
            Watch(bridgeActor);
            // Resolve the snapshot so the session is fully established.
            bridgeActor.Tell(new ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView.DebugViewSnapshot(
                "Site1.Pump01",
                new List<ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming.AttributeValueChanged>(),
                new List<ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming.AlarmStateChanged>(),
                DateTimeOffset.UtcNow));
            var session = await startTask;
            // Several orphan windows with NO traffic of any kind — only the service's
            // keepalive timer. The session must survive.
            await Task.Delay(TimeSpan.FromMilliseconds(1500));
            ExpectNoMsg(TimeSpan.FromMilliseconds(100));
            Assert.False(bridgeActor.IsNobody());
            // Detach the consumer: the session leaves the registry, keepalives stop, and the
            // actor is stopped (StopStream) — the orphan net is the backstop for the case
            // where that explicit stop never happens.
            service.StopStream(session.SessionId);
            ExpectTerminated(bridgeActor, TimeSpan.FromSeconds(3));
        }
        finally
        {
            DebugStreamService.KeepaliveInterval = previousKeepalive;
            DebugStreamBridgeActor.ConsumerIdleTimeout = previousIdle;
            DebugStreamBridgeActor.ConsumerLivenessCheckInterval = previousCheck;
        }
    }
 }
@@ -11,6 +11,13 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;
 /// <summary>
 /// Tests for DebugStreamBridgeActor with gRPC streaming integration.
 /// </summary>
 /// <remarks>
 /// Shares the <c>DebugStreamStatics</c> xUnit collection with <c>DebugStreamServiceTests</c>:
 /// both tune the actor's static test seams (<c>SnapshotTimeout</c>, <c>ConsumerIdleTimeout</c>,
 /// …), and xUnit parallelizes distinct classes by default, which would let one class's
 /// try/finally restore clobber the other's window mid-test.
 /// </remarks>
 [Collection("DebugStreamStatics")]
 public class DebugStreamBridgeActorTests : TestKit
 {
    private const string SiteId = "site-alpha";
@@ -1034,14 +1041,11 @@ public class DebugStreamBridgeActorTests : TestKit
    }
    [Fact]
-    public void StreamEvents_AreWrapped_SoTheyDoNotResetTheOrphanReceiveTimeout()
+    public void StreamEvents_StillReachTheConsumer_ThroughTheWrappedCallbackPath()
    {
-        // Structural pin for the fix: the gRPC callback wraps every event in an envelope
+        // The gRPC callback wraps every event in LiveDebugStreamEvent; the wrapper must be
-        // marked INotInfluenceReceiveTimeout, so a busy site can no longer keep an
+        // transparent to delivery (it exists only to keep the high-volume path explicit —
-        // abandoned session alive indefinitely by feeding it events.
+        // the orphan net keys off the consumer keepalive, not the mailbox).
        Assert.True(typeof(Akka.Actor.INotInfluenceReceiveTimeout)
            .IsAssignableFrom(typeof(LiveDebugStreamEvent)));
        var ctx = CreateBridgeActor();
        ctx.CommProbe.ExpectMsg<SiteEnvelope>();
        AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3));
@@ -1052,8 +1056,6 @@ public class DebugStreamBridgeActorTests : TestKit
            new List<AlarmStateChanged>(),
            DateTimeOffset.UtcNow));
        // An event delivered through the real gRPC callback path still reaches the
        // consumer — the wrapper is transparent to delivery.
        var evt = new AttributeValueChanged(
            InstanceName, "Modules.IO", "Temperature", 42.5, "Good", DateTimeOffset.UtcNow);
        ctx.MockGrpcClient.SubscribeCalls[0].OnEvent(evt);
@@ -1063,6 +1065,134 @@ public class DebugStreamBridgeActorTests : TestKit
            lock (ctx.ReceivedEvents) { return ctx.ReceivedEvents.OfType<AttributeValueChanged>().Any(); }
        }, TimeSpan.FromSeconds(3));
    }
    // ----- Orphan net: measures the CONSUMER, not the mailbox ----- //
    [Fact]
    public void HealthySession_StreamingEvents_WithConsumerKeepalives_SurvivesWellPastTheOrphanWindow()
    {
        // THE regression this closes: with the orphan net armed off the mailbox
        // (SetReceiveTimeout) and stream events correctly excluded from it, nothing recurring
        // reset it — the snapshot lands once, GrpcStreamStable once — so a perfectly healthy
        // session self-terminated one window later and the consumer was told
        // "Site disconnected". Here the session streams events and receives the keepalive
        // DebugStreamService sends while it is attached; it must live through MANY windows.
        DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMilliseconds(400);
        DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromMilliseconds(50);
        try
        {
            var ctx = CreateBridgeActor();
            ctx.CommProbe.ExpectMsg<SiteEnvelope>();
            AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3));
            Watch(ctx.BridgeActor);
            ctx.BridgeActor.Tell(new DebugViewSnapshot(
                InstanceName,
                new List<AttributeValueChanged>(),
                new List<AlarmStateChanged>(),
                DateTimeOffset.UtcNow));
            // ~5 orphan windows of pure stream traffic + consumer keepalives, and no other
            // mailbox activity whatsoever (no reconnects, no snapshots, no stop).
            var deadline = DateTime.UtcNow.AddSeconds(2);
            var delivered = 0;
            while (DateTime.UtcNow < deadline)
            {
                ctx.MockGrpcClient.SubscribeCalls[0].OnEvent(new AttributeValueChanged(
                    InstanceName, "Modules.IO", "Temperature", 20.0 + delivered, "Good",
                    DateTimeOffset.UtcNow));
                delivered++;
                // What DebugStreamService's shared timer does for an attached session.
                ctx.BridgeActor.Tell(new DebugStreamConsumerAlive());
                Thread.Sleep(100);
            }
            ExpectNoMsg(TimeSpan.FromMilliseconds(100));
            Assert.False(ctx.TerminatedFlag[0]);
            // Still serving: the events all arrived and the actor is alive.
            AwaitCondition(() =>
            {
                lock (ctx.ReceivedEvents) { return ctx.ReceivedEvents.OfType<AttributeValueChanged>().Count() == delivered; }
            }, TimeSpan.FromSeconds(3));
            Assert.False(ctx.BridgeActor.IsNobody());
        }
        finally
        {
            DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMinutes(5);
            DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromSeconds(30);
        }
    }
    [Fact]
    public void OrphanedSession_ConsumerGone_StillTerminates_EvenWhileEventsKeepArriving()
    {
        // The other half of the contract: site chatter must NOT hold an abandoned session
        // open. No keepalive arrives (the consumer is gone), so the session terminates,
        // unsubscribes from the site and reports termination — while events stream in.
        DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMilliseconds(400);
        DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromMilliseconds(50);
        try
        {
            var ctx = CreateBridgeActor();
            ctx.CommProbe.ExpectMsg<SiteEnvelope>();
            AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3));
            Watch(ctx.BridgeActor);
            ctx.BridgeActor.Tell(new DebugViewSnapshot(
                InstanceName,
                new List<AttributeValueChanged>(),
                new List<AlarmStateChanged>(),
                DateTimeOffset.UtcNow));
            // Keep the site chatty for longer than the orphan window — with no keepalive.
            var deadline = DateTime.UtcNow.AddSeconds(1);
            while (DateTime.UtcNow < deadline)
            {
                ctx.MockGrpcClient.SubscribeCalls[0].OnEvent(new AttributeValueChanged(
                    InstanceName, "Modules.IO", "Temperature", 42.5, "Good", DateTimeOffset.UtcNow));
                Thread.Sleep(50);
            }
            ExpectTerminated(ctx.BridgeActor, TimeSpan.FromSeconds(3));
            Assert.True(ctx.TerminatedFlag[0]);
            Assert.Contains("corr-1", ctx.MockGrpcClient.UnsubscribedCorrelationIds);
        }
        finally
        {
            DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMinutes(5);
            DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromSeconds(30);
        }
    }
    [Fact]
    public void ConsumerKeepalive_RenewsTheWindow_AfterANearMiss()
    {
        // A single late keepalive is enough to save a session — the stamp is refreshed, not
        // a one-shot grace period.
        DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMilliseconds(500);
        DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromMilliseconds(50);
        try
        {
            var ctx = CreateBridgeActor();
            ctx.CommProbe.ExpectMsg<SiteEnvelope>();
            Watch(ctx.BridgeActor);
            Thread.Sleep(350);                                   // most of the window burnt
            ctx.BridgeActor.Tell(new DebugStreamConsumerAlive()); // …then the consumer checks in
            Thread.Sleep(350);                                   // past the ORIGINAL deadline
            Assert.False(ctx.TerminatedFlag[0]);
            // …and once the keepalives stop, it does terminate.
            ExpectTerminated(ctx.BridgeActor, TimeSpan.FromSeconds(3));
        }
        finally
        {
            DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMinutes(5);
            DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromSeconds(30);
        }
    }
 }
 /// <summary>
@@ -720,6 +720,99 @@ public class SiteAlarmAggregatorActorTests : TestKit
        AwaitCondition(() => !sink.StreamLive, TimeSpan.FromSeconds(3));
    }
    [Fact]
    public void ConsecutiveReconcileTicks_WithNoReconnect_EachRunAFanout()
    {
        // The bug: a tick's OWN fan-out completion armed the skip flag, so steady state ran
        // fan-out → skip → fan-out → skip — one reconcile per TWO intervals, halving both the
        // not-reporting refresh and the alarm reconcile backstop. Nothing was being
        // duplicated: only a connect/failover-driven seed can make a tick redundant.
        var (actor, seed, _, _) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        seed.CompleteNext();                 // initial seed — this one DOES stand a tick down
        Thread.Sleep(150);
        actor.Tell(new RunReconcile());      // consumed by the initial seed's flag
        Thread.Sleep(200);
        Assert.Equal(1, seed.CallCount);
        // From here on, with no reconnect in between, EVERY tick must fan out.
        for (var expected = 2; expected <= 5; expected++)
        {
            actor.Tell(new RunReconcile());
            AwaitCondition(() => seed.CallCount == expected, TimeSpan.FromSeconds(3));
            seed.CompleteNext();
            Thread.Sleep(100);
        }
        Assert.Equal(5, seed.CallCount);
    }
    [Fact]
    public void ConnectDrivenReseed_StillStandsTheNextTickDown()
    {
        // The other half of the fix: the skip exists for the connect/failover seed, and that
        // suppression must survive. A reconnect's re-seed still makes the very next tick a
        // no-op, so a flapping stream cannot double the whole-site snapshot rate.
        var (actor, seed, _, factory) = CreateActor(
            reconcileInterval: TimeSpan.FromMinutes(10), autoConnect: false);
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        factory.ClientFor(GrpcNodeA).Subs[0].OnConnected!();
        seed.CompleteNext();
        Thread.Sleep(150);
        actor.Tell(new RunReconcile());   // consumed by the initial seed
        actor.Tell(new RunReconcile());   // real tick fan-out
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
        seed.CompleteNext();
        Thread.Sleep(150);
        // Stream faults → reconnect on node B → connect-driven re-seed (call 3).
        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("site gone"));
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeB).Subs[0].OnConnected!();
        AwaitCondition(() => seed.CallCount == 3, TimeSpan.FromSeconds(3));
        seed.CompleteNext();
        Thread.Sleep(150);
        // That re-seed covers this window: the next tick is skipped…
        actor.Tell(new RunReconcile());
        Thread.Sleep(250);
        Assert.Equal(3, seed.CallCount);
        // …and the one after it fans out again.
        actor.Tell(new RunReconcile());
        AwaitCondition(() => seed.CallCount == 4, TimeSpan.FromSeconds(3));
    }
    [Fact]
    public void ForeignCancelledStreamError_DropsLiveness_AndReopensOnTheOtherNode()
    {
        // End-to-end for the SiteStreamGrpcClient fix: a peer-originated
        // RpcException(Cancelled) now reaches onError instead of being swallowed. Before the
        // fix NONE of onError/onCompleted/onConnected fired, so the aggregator kept
        // _streamDown=false — IsLive stuck true and the reconcile tick's reopen guard, which
        // only fires when the stream is known down, never ran.
        var (_, seed, sink, factory) = CreateActor(
            reconcileInterval: TimeSpan.FromMinutes(10), autoConnect: false);
        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
        factory.ClientFor(GrpcNodeA).Subs[0].OnConnected!();
        seed.CompleteNext();
        AwaitCondition(() => sink.StreamLive, TimeSpan.FromSeconds(3));
        factory.ClientFor(GrpcNodeA).Subs[0].OnError(
            new global::Grpc.Core.RpcException(new global::Grpc.Core.Status(
                global::Grpc.Core.StatusCode.Cancelled, "cancelled by peer")));
        AwaitCondition(() => !sink.StreamLive, TimeSpan.FromSeconds(3));
        // …and the stream is reopened on the other node, where a connect earns one re-seed.
        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
        factory.ClientFor(GrpcNodeB).Subs[0].OnConnected!();
        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
    }
    [Fact]
    public void FailedSeed_IsRetried_WithBackoff_BeforeTheNextReconcile()
    {
@@ -351,9 +351,170 @@ public class SiteStreamGrpcClientTests
        Assert.Equal(0, completed);
    }
    // --- Foreign vs. own Cancelled (review F2) ---
    [Fact]
    public async Task ConsumeStream_ForeignCancelled_InvokesOnError()
    {
        // A Cancelled status we did NOT ask for — the PEER cancelled, or the channel was
        // disposed underneath us. It used to be swallowed by an unguarded
        // `when (ex.StatusCode == StatusCode.Cancelled)` filter, firing none of
        // onError/onCompleted/onConnected: the consuming aggregator kept _streamDown=false,
        // so IsLive stayed true forever and the reconcile tick's reopen guard never fired.
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource(); // deliberately NOT cancelled
        Exception? error = null;
        var completed = 0;
        await client.ConsumeStreamAsync(
            "corr-foreign-cancel",
            cts,
            () => FakeCall(new StubStreamReader(
                new RpcException(new Status(StatusCode.Cancelled, "cancelled by peer")))),
            _ => { },
            ex => error = ex,
            () => completed++);
        var rpc = Assert.IsType<RpcException>(error);
        Assert.Equal(StatusCode.Cancelled, rpc.StatusCode);
        Assert.Equal(0, completed);
        Assert.False(cts.IsCancellationRequested);
    }
    [Fact]
    public async Task ConsumeStream_OwnCancelledRpcException_InvokesNeitherCallback()
    {
        // The other side of the same filter: OUR cancellation surfacing as
        // RpcException(Cancelled) is a teardown, so neither callback fires.
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource();
        await cts.CancelAsync();
        Exception? error = null;
        var completed = 0;
        await client.ConsumeStreamAsync(
            "corr-own-cancel",
            cts,
            () => throw new RpcException(new Status(StatusCode.Cancelled, "we cancelled")),
            _ => { },
            ex => error = ex,
            () => completed++);
        Assert.Null(error);
        Assert.Equal(0, completed);
    }
    // --- onConnected is only ever raised on real proof of a live peer (review F3) ---
    [Fact]
    public async Task ConsumeStream_HeadersArrive_InvokesOnConnectedOnce()
    {
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource();
        var connected = 0;
        await client.ConsumeStreamAsync(
            "corr-headers",
            cts,
            () => FakeCall(
                new StubStreamReader(
                    new SiteStreamEvent { CorrelationId = "corr-headers" },
                    new SiteStreamEvent { CorrelationId = "corr-headers" }),
                Task.FromResult(new Metadata())),
            _ => { },
            _ => { },
            () => { },
            () => connected++);
        // Once — the two events must not re-raise it.
        Assert.Equal(1, connected);
    }
    [Fact]
    public async Task ConsumeStream_HeaderTimeoutOnADeadSite_NeverReportsConnected()
    {
        // The bug: a header TIMEOUT used to be reported as connected. An unreachable/wedged
        // site produces exactly that shape, so the aggregator cleared _streamDown, consumed
        // its pending re-seed and fanned a full snapshot out at a site that never answered.
        var previous = SiteStreamGrpcClient.ConnectedHeaderTimeout;
        SiteStreamGrpcClient.ConnectedHeaderTimeout = TimeSpan.FromMilliseconds(50);
        try
        {
            var client = SiteStreamGrpcClient.CreateForTesting();
            var cts = new CancellationTokenSource();
            var connected = 0;
            var neverArrives = new TaskCompletionSource<Metadata>();
            // No headers AND no events: the site is dead. The stream ends (status OK) with
            // no connected signal ever raised.
            await client.ConsumeStreamAsync(
                "corr-dead",
                cts,
                () => FakeCall(new StubStreamReader(), neverArrives.Task),
                _ => { },
                _ => { },
                () => { },
                () => connected++);
            Assert.Equal(0, connected);
        }
        finally
        {
            SiteStreamGrpcClient.ConnectedHeaderTimeout = previous;
        }
    }
    [Fact]
    public async Task ConsumeStream_HeaderTimeoutThenEvent_ReportsConnectedOnceFromTheEvent()
    {
        // The case the timeout was originally added for — a peer that defers its headers
        // until the first message — is now covered by the event itself, which is real proof
        // of a live peer. Connected must precede the event's delivery and fire only once.
        var previous = SiteStreamGrpcClient.ConnectedHeaderTimeout;
        SiteStreamGrpcClient.ConnectedHeaderTimeout = TimeSpan.FromMilliseconds(50);
        try
        {
            var client = SiteStreamGrpcClient.CreateForTesting();
            var cts = new CancellationTokenSource();
            var connected = 0;
            var connectedBeforeFirstEvent = false;
            var events = 0;
            var neverArrives = new TaskCompletionSource<Metadata>();
            await client.ConsumeStreamAsync(
                "corr-late-headers",
                cts,
                () => FakeCall(
                    new StubStreamReader(
                        new SiteStreamEvent { CorrelationId = "corr-late-headers" },
                        new SiteStreamEvent { CorrelationId = "corr-late-headers" }),
                    neverArrives.Task),
                _ =>
                {
                    if (events == 0) connectedBeforeFirstEvent = connected == 1;
                    events++;
                },
                _ => { },
                () => { },
                () => connected++);
            Assert.Equal(2, events);
            Assert.Equal(1, connected);
            Assert.True(connectedBeforeFirstEvent);
        }
        finally
        {
            SiteStreamGrpcClient.ConnectedHeaderTimeout = previous;
        }
    }
    private static AsyncServerStreamingCall<SiteStreamEvent> FakeCall(StubStreamReader reader) =>
        FakeCall(reader, Task.FromResult(new Metadata()));
    private static AsyncServerStreamingCall<SiteStreamEvent> FakeCall(
        StubStreamReader reader, Task<Metadata> responseHeaders) =>
        new(reader,
-            Task.FromResult(new Metadata()),
+            responseHeaders,
            () => Status.DefaultSuccess,
            () => new Metadata(),
            () => { });
@@ -243,6 +243,45 @@ public class SiteStreamPullAuditEventsTests : TestKit
        Assert.True(response.MoreAvailable);
    }
    [Fact]
    public async Task PullAuditEvents_RetiresBeforeItServes_SoThisBatchIsNeverSelfRetired()
    {
        // Ordering is load-bearing for the "only served rows retire" invariant. The queue
        // bounds the cursor flip by insertion order — the high-water mark of rows it has
        // SERVED — so the retire step must run BEFORE the read: retiring first can only ever
        // reach rows served by an EARLIER pull, never the ones this call is about to serve
        // (which would defeat at-least-once), and never a row inserted after them (the
        // late-stamped insert that used to be silently retired and then age-purged).
        var queue = Substitute.For<ISiteAuditQueue>();
        queue.ReadPendingSinceAsync(
                Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
            .Returns((IReadOnlyList<AuditEvent>)new[] { NewEvent() });
        var server = CreateServer();
        server.SetSiteAuditQueue(queue);
        var cursorTime = DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc);
        await server.PullAuditEvents(
            new PullAuditEventsRequest
            {
                SinceUtc = Timestamp.FromDateTime(cursorTime),
                BatchSize = 100,
            },
            NewContext());
        Received.InOrder(() =>
        {
            queue.MarkReconciledUpToAsync(cursorTime, null, Arg.Any<CancellationToken>());
            queue.ReadPendingSinceAsync(cursorTime, 100, null, Arg.Any<CancellationToken>());
        });
        // Serving still flips nothing by itself — the next cursor is the only receipt.
        await queue.DidNotReceive().MarkReconciledAsync(
            Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
        await queue.DidNotReceive().MarkForwardedAsync(
            Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
    }
    [Fact]
    public async Task PullAuditEvents_MarkReconciledUpToThrows_ResponseStillReturned()
    {