Merge branch 'worktree-agent-a143c0cc0b4d07e76' into arch-review-remediation

2026-08-14 20:14:13 -04:00
parent 0b201e410c 34a3f4bb69
commit 87e25c790c
9 changed files with 507 additions and 45 deletions
@@ -343,6 +343,7 @@ Disconnect is detected at the **transport layer**, never via an application-leve

 - **In-flight command/control messages (gRPC call + deadline)**: When a connection drops while a request is in flight (e.g., a deployment sent but no response received), the gRPC call fails or hits its deadline and the caller receives a failure. There is **no automatic retry or buffering at central** — the engineer sees the failure in the UI and re-initiates the action. This is consistent with the design principle that central does not buffer messages. An in-progress deployment whose round-trip exceeds the timeout (default 120 s at `CommunicationService.DeployInstanceAsync`) surfaces as `DeploymentStatus.Failed` to the caller.
 - **Debug streams (gRPC)**: Any gRPC stream interruption is detected by the HTTP/2 keepalive PING (~25 s) and triggers reconnection logic in the `DebugStreamBridgeActor`. The bridge actor attempts to reconnect to the other site node endpoint (NodeB if NodeA failed, or vice versa), with up to 3 retries and 5-second backoff. If all retries fail, the consumer is notified via `OnStreamTerminated` and the bridge actor is stopped. Events during the reconnection gap are lost (acceptable for real-time debug view). On successful reconnection, the consumer can request a fresh snapshot to re-sync state.
+- **Graceful end of stream is NOT a fault, but it IS a reconnect trigger.** When the site's `GrpcMaxStreamLifetime` (4 h) elapses — and on a graceful site shutdown — the server ends the RPC with status **OK**, so the client's read loop simply finishes with no exception. `SiteStreamGrpcClient` reports that through a dedicated `onCompleted` callback, distinct from `onError`, and both consuming actors treat it as a reconnect trigger that **does not spend (or refund) the error-retry budget** and does **not** flip the node — the peer that closed cleanly is healthy. `SiteAlarmAggregatorActor` marks the stream down and lets the periodic reconcile tick reopen it (which also re-seeds); `DebugStreamBridgeActor` reopens on the same endpoint after `ReconnectDelay`. Both stamp the stream generation on the callback so a completion racing out of an already-replaced stream is ignored, and both observe the subscription `Task` itself so a fault escaping the subscribe call can never go unobserved. Without the `onCompleted` leg an OK completion was invisible to the reconnect logic and the stream stayed silently dead until the central process restarted.

 ## Failover Behavior

@@ -63,6 +63,14 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
    private bool _stopped;
    private CancellationTokenSource? _grpcCts;

+    /// <summary>
+    /// Monotonic stream generation stamped on each opened gRPC stream and echoed back on its
+    /// error/completion callbacks: a late callback raced out of a previous (cancelled) stream
+    /// carries a stale generation and is ignored, so it can neither burn retry budget nor
+    /// open a duplicate stream. Mirrors <c>SiteAlarmAggregatorActor</c>. Actor-thread only.
+    /// </summary>
+    private int _streamGeneration;
+
    /// <summary>
    /// Phase flag. <see langword="false"/> until the initial
    /// <see cref="DebugViewSnapshot"/> has been delivered and the pre-snapshot buffer
@@ -198,10 +206,33 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
        // gRPC stream error — attempt reconnection
        Receive<GrpcStreamError>(msg =>
        {
+            // Ignore a late error raced out of a previous (cancelled) stream: it must not
+            // burn retry budget or flip the node a second time.
+            if (msg.Generation != _streamGeneration)
+            {
+                _log.Debug("Ignoring stale gRPC error from stream generation {0} (current {1})",
+                    msg.Generation, _streamGeneration);
+                return;
+            }
            _log.Warning("gRPC stream error for {0}: {1}", _instanceUniqueName, msg.Exception.Message);
            HandleGrpcError();
        });

+        // gRPC stream ended GRACEFULLY (server status OK) — the site's 4h max stream
+        // lifetime elapsing or a graceful site shutdown. Not a fault: reopen on the SAME
+        // node without spending retry budget. Without this the session went silently deaf.
+        Receive<GrpcStreamCompleted>(msg =>
+        {
+            if (_stopped) return;
+            if (msg.Generation != _streamGeneration)
+            {
+                _log.Debug("Ignoring stale gRPC completion from stream generation {0} (current {1})",
+                    msg.Generation, _streamGeneration);
+                return;
+            }
+            HandleGrpcCompleted();
+        });
+
        // Scheduled reconnection
        Receive<ReconnectGrpcStream>(_ => OpenGrpcStream());

@@ -424,20 +455,57 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
        // retry budget is recovered. Cancelled by HandleGrpcError.
        Timers.StartSingleTimer(StabilityTimerKey, new GrpcStreamStable(), StabilityWindow);

+        var generation = ++_streamGeneration;
        var client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint);
        var self = Self;
        var ct = _grpcCts.Token;

-        // Launch as background task — onEvent and onError marshal back to actor via Tell
+        // Launch as background task — the callbacks marshal back to the actor via Tell.
+        // The task itself is observed below: a fault escaping SubscribeAsync would otherwise
+        // leave the session waiting on a stream that does not exist, exception unobserved.
        Task.Run(async () =>
        {
            await client.SubscribeAsync(
                _correlationId,
                _instanceUniqueName,
                evt => self.Tell(evt),
-                ex => self.Tell(new GrpcStreamError(ex)),
+                ex => self.Tell(new GrpcStreamError(ex, generation)),
+                () => self.Tell(new GrpcStreamCompleted(generation)),
                ct);
-        }, ct);
+        }, ct).ContinueWith(t =>
+        {
+            if (t.IsFaulted)
+                self.Tell(new GrpcStreamError(t.Exception!.GetBaseException(), generation));
+            else if (t.IsCanceled && !ct.IsCancellationRequested)
+                self.Tell(new GrpcStreamCompleted(generation));
+            // RanToCompletion: SubscribeAsync already reported its own outcome.
+        }, TaskContinuationOptions.ExecuteSynchronously);
+    }
+
+    /// <summary>
+    /// Handles a graceful end of stream (server status OK). The stream simply expired or the
+    /// site shut down cleanly, so the retry budget is left untouched and the endpoint is not
+    /// flipped; the reopen is scheduled through the existing reconnect timer, which also
+    /// rate-limits a pathological site that keeps closing streams immediately.
+    /// </summary>
+    private void HandleGrpcCompleted()
+    {
+        // The stream is gone, so its armed stability timer must not later "recover" a
+        // budget that its successor has since spent.
+        Timers.Cancel(StabilityTimerKey);
+
+        _log.Info("gRPC stream for {0} completed gracefully (server end of stream); reopening",
+            _instanceUniqueName);
+
+        // Release the site-side relay for the finished stream before reopening, so the site
+        // is not left with a zombie relay actor for this correlation id.
+        _grpcCts?.Cancel();
+        _grpcCts?.Dispose();
+        _grpcCts = null;
+        _grpcFactory.TryGet(_siteIdentifier, _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress)
+            ?.Unsubscribe(_correlationId);
+
+        Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectGrpcStream(), ReconnectDelay);
    }

    private void HandleGrpcError()
@@ -512,9 +580,17 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
 public record StopDebugStream;

 /// <summary>
-/// Internal message indicating a gRPC stream error occurred.
+/// Internal message indicating a gRPC stream error occurred, stamped with the stream
+/// generation it came from so a late error out of a cancelled stream can be ignored.
 /// </summary>
-internal record GrpcStreamError(Exception Exception);
+internal record GrpcStreamError(Exception Exception, int Generation);
+
+/// <summary>
+/// Internal message indicating the gRPC stream ended gracefully (server status OK — the
+/// site's max stream lifetime elapsed, or the site shut down cleanly), stamped with its
+/// stream generation.
+/// </summary>
+internal record GrpcStreamCompleted(int Generation);

 /// <summary>
 /// Internal message to trigger gRPC stream reconnection.
@@ -78,12 +78,23 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
    private bool _stopped;

    /// <summary>
-    /// True once the live gRPC stream has been given up (retry budget exhausted). Reconcile
-    /// snapshots keep serving in the meantime; the next reconcile tick self-heals the stream
-    /// by resetting the retry budget and reopening it, so a sustained site outage does not
-    /// permanently drop the live feed. Actor-thread only.
+    /// True while there is no live gRPC stream — either it was given up (retry budget
+    /// exhausted) or it ended gracefully (server status OK at the site's max stream
+    /// lifetime). Reconcile snapshots keep serving in the meantime; the next reconcile tick
+    /// self-heals the stream by resetting the retry budget and reopening it, so neither a
+    /// sustained site outage nor a routine 4h stream expiry permanently drops the live feed.
+    /// Actor-thread only.
    /// </summary>
    private bool _streamDown;
+
+    /// <summary>
+    /// Why the stream is down: <c>true</c> = the retry budget was exhausted, so the
+    /// self-healing reopen must also reset it; <c>false</c> = it ended gracefully, and the
+    /// budget — which a completion neither spends nor refunds — is carried across the reopen
+    /// untouched (a stream flapping between faults and clean closes must still trip
+    /// MaxRetries). Actor-thread only.
+    /// </summary>
+    private bool _retryBudgetExhausted;
    private CancellationTokenSource? _grpcCts;
    private CancellationTokenSource? _lifetimeCts;

@@ -218,6 +229,22 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
            HandleGrpcError();
        });

+        // gRPC stream ended GRACEFULLY (server status OK) — the site's 4h max stream
+        // lifetime elapsing or a graceful site shutdown. Not a fault: the stream is marked
+        // down so the reconcile tick reopens it (which also re-seeds), but the retry budget
+        // is untouched and the node is not flipped. Same generation fence as the error path.
+        Receive<GrpcAlarmStreamCompleted>(msg =>
+        {
+            if (_stopped) return;
+            if (msg.Generation != _streamGeneration)
+            {
+                _log.Debug("Ignoring stale gRPC completion from stream generation {0} (current {1})",
+                    msg.Generation, _streamGeneration);
+                return;
+            }
+            HandleGrpcCompleted();
+        });
+
        Receive<ReconnectAlarmStream>(_ => OpenGrpcStream());

        // Owning service asks us to stop (last viewer left + linger elapsed).
@@ -279,7 +306,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        if (_streamDown)
        {
            _log.Info("Site-alarm gRPC stream for {0} was down; reopening on reconcile tick", _siteIdentifier);
+            if (_retryBudgetExhausted)
+            {
+                _retryBudgetExhausted = false;
                _retryCount = 0;
+            }
            // Telemetry: a reconcile-driven reopen after the stream was given up is a reconnect.
            ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();
            OpenGrpcStream();
@@ -514,14 +545,26 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        var self = Self;
        var ct = _grpcCts.Token;

+        // The subscription task itself is observed (below): a fault escaping
+        // SubscribeSiteAsync — or a Task.Run that never started because ct was already
+        // cancelled — would otherwise leave the actor waiting on a stream that does not
+        // exist, with the exception silently unobserved.
        Task.Run(async () =>
        {
            await client.SubscribeSiteAsync(
                _correlationId,
                alarm => self.Tell(alarm),
                ex => self.Tell(new GrpcAlarmStreamError(ex, generation)),
+                () => self.Tell(new GrpcAlarmStreamCompleted(generation)),
                ct);
-        }, ct);
+        }, ct).ContinueWith(t =>
+        {
+            if (t.IsFaulted)
+                self.Tell(new GrpcAlarmStreamError(t.Exception!.GetBaseException(), generation));
+            else if (t.IsCanceled && !ct.IsCancellationRequested)
+                self.Tell(new GrpcAlarmStreamCompleted(generation));
+            // RanToCompletion: SubscribeSiteAsync already reported its own outcome.
+        }, TaskContinuationOptions.ExecuteSynchronously);
    }

    private void HandleGrpcError()
@@ -543,6 +586,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
                       "reconcile snapshots continue and the next reconcile tick will retry the stream",
                       _siteIdentifier, MaxRetries);
            _streamDown = true;
+            _retryBudgetExhausted = true;
            CleanupGrpc();
            return;
        }
@@ -569,6 +613,26 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
            Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectAlarmStream(), _reconnectDelay);
    }

+    /// <summary>
+    /// Handles a graceful end of stream. The stream is torn down and left down for the
+    /// reconcile tick to reopen — deliberately NOT reopened inline, so a site that keeps
+    /// completing streams immediately can never spin this actor into a hot reconnect loop.
+    /// The retry budget is neither spent nor reset here (completion is not a fault), and the
+    /// endpoint is not flipped: the node that just closed a stream cleanly is healthy.
+    /// </summary>
+    private void HandleGrpcCompleted()
+    {
+        // The stream is gone, so its armed stability timer must not later "recover" a
+        // budget that its successor has since spent.
+        Timers.Cancel(StabilityTimerKey);
+
+        _log.Info("Site-alarm gRPC stream for {0} completed gracefully (server end of stream); " +
+                  "reopening on the next reconcile tick", _siteIdentifier);
+
+        _streamDown = true;
+        CleanupGrpc();
+    }
+
    private void CleanupGrpc()
    {
        _grpcCts?.Cancel();
@@ -619,6 +683,11 @@ internal sealed record PublishCoalesced;
 /// generation it came from so a late error from a cancelled stream can be ignored (N7.2).</summary>
 internal sealed record GrpcAlarmStreamError(Exception Exception, int Generation);

+/// <summary>Internal: the site-alarm gRPC stream ended gracefully (server status OK — max
+/// stream lifetime or site shutdown), stamped with its stream generation so a late completion
+/// from a cancelled stream can be ignored.</summary>
+internal sealed record GrpcAlarmStreamCompleted(int Generation);
+
 /// <summary>Internal: reconnect the site-alarm gRPC stream (flip node).</summary>
 internal sealed record ReconnectAlarmStream;

@@ -167,13 +167,19 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
    /// <summary>
    /// Opens a server-streaming subscription for a specific instance.
    /// This is a long-running async method; the caller launches it as a background task.
-    /// The <paramref name="onEvent"/> callback delivers domain events, and
-    /// <paramref name="onError"/> lets the caller handle reconnection.
+    /// The <paramref name="onEvent"/> callback delivers domain events,
+    /// <paramref name="onError"/> lets the caller handle reconnection after a fault, and
+    /// <paramref name="onCompleted"/> reports a graceful end of stream.
    /// </summary>
    /// <param name="correlationId">Unique identifier for this subscription.</param>
    /// <param name="instanceUniqueName">Unique name of the instance to subscribe to.</param>
    /// <param name="onEvent">Callback invoked for each domain event received from the stream.</param>
    /// <param name="onError">Callback invoked when the subscription encounters an error.</param>
+    /// <param name="onCompleted">
+    /// Callback invoked when the server ended the RPC with OK — the site's max stream
+    /// lifetime elapsing or a graceful site shutdown. Mutually exclusive with
+    /// <paramref name="onError"/>; see <see cref="ConsumeStreamAsync"/>.
+    /// </param>
    /// <param name="ct">Cancellation token to stop the subscription.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public virtual async Task SubscribeAsync(
@@ -181,6 +187,7 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
        string instanceUniqueName,
        Action<object> onEvent,
        Action<Exception> onError,
+        Action onCompleted,
        CancellationToken ct)
    {
        if (_client is null)
@@ -195,30 +202,18 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
            InstanceUniqueName = instanceUniqueName
        };

-        try
-        {
-            using var call = _client.SubscribeInstance(request, cancellationToken: cts.Token);
-
-            await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token))
+        await ConsumeStreamAsync(
+            correlationId,
+            cts,
+            () => _client.SubscribeInstance(request, cancellationToken: cts.Token),
+            evt =>
            {
                var domainEvent = ConvertToDomainEvent(evt);
                if (domainEvent != null)
                    onEvent(domainEvent);
-            }
-        }
-        catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
-        {
-            // Normal cancellation — not an error
-        }
-        catch (Exception ex)
-        {
-            onError(ex);
-        }
-        finally
-        {
-            // Remove only our own entry -- a racing reconnect may already own the slot.
-            RemoveSubscription(correlationId, cts);
-        }
+            },
+            onError,
+            onCompleted);
    }

    /// <summary>
@@ -244,12 +239,18 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
    /// <param name="correlationId">Unique identifier for this subscription.</param>
    /// <param name="onAlarmEvent">Callback invoked for each alarm delta received from the site-wide stream.</param>
    /// <param name="onError">Callback invoked when the subscription encounters an error.</param>
+    /// <param name="onCompleted">
+    /// Callback invoked when the server ended the RPC with OK — the site's max stream
+    /// lifetime elapsing or a graceful site shutdown. Mutually exclusive with
+    /// <paramref name="onError"/>; see <see cref="ConsumeStreamAsync"/>.
+    /// </param>
    /// <param name="ct">Cancellation token to stop the subscription.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public virtual async Task SubscribeSiteAsync(
        string correlationId,
        Action<AlarmStateChanged> onAlarmEvent,
        Action<Exception> onError,
+        Action onCompleted,
        CancellationToken ct)
    {
        if (_client is null)
@@ -263,21 +264,73 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
            CorrelationId = correlationId
        };

-        try
-        {
-            using var call = _client.SubscribeSite(request, cancellationToken: cts.Token);
-
-            await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token))
+        await ConsumeStreamAsync(
+            correlationId,
+            cts,
+            () => _client.SubscribeSite(request, cancellationToken: cts.Token),
+            evt =>
            {
                // Site-wide stream is alarm-only by contract; defensively ignore anything else.
                if (ConvertToAlarmEvent(evt) is { } alarm)
                    onAlarmEvent(alarm);
+            },
+            onError,
+            onCompleted);
    }
+
+    /// <summary>
+    /// Drains a server-streaming call to its end, classifying the outcome into exactly one
+    /// of three terminations: a fault (<paramref name="onError"/>), a graceful server-side
+    /// end of stream (<paramref name="onCompleted"/>), or our own cancellation (neither).
+    /// <para>
+    /// The graceful case is the one that matters operationally: the site server caps every
+    /// stream at <c>CommunicationOptions.GrpcMaxStreamLifetime</c> (4 h) and, when that
+    /// elapses, ends the RPC with status OK. The client-side <c>await foreach</c> then simply
+    /// finishes, so before <c>onCompleted</c> existed the consuming actor observed nothing at
+    /// all and the stream stayed silently dead until the process restarted.
+    /// </para>
+    /// Internal so the completion/fault classification can be unit-tested with a fake
+    /// <see cref="IAsyncStreamReader{T}"/> and no live channel.
+    /// </summary>
+    /// <param name="correlationId">Unique identifier for this subscription.</param>
+    /// <param name="cts">The subscription's linked token source (owned by the caller).</param>
+    /// <param name="openCall">
+    /// Opens the server-streaming call. Invoked inside the guarded region so a throw while
+    /// opening is reported through <paramref name="onError"/> like any other stream fault.
+    /// </param>
+    /// <param name="onEvent">Invoked per wire event.</param>
+    /// <param name="onError">Invoked once if the stream faulted.</param>
+    /// <param name="onCompleted">Invoked once if the server ended the stream with OK.</param>
+    /// <returns>A task that completes when the stream has ended and its outcome been reported.</returns>
+    internal async Task ConsumeStreamAsync(
+        string correlationId,
+        CancellationTokenSource cts,
+        Func<AsyncServerStreamingCall<SiteStreamEvent>> openCall,
+        Action<SiteStreamEvent> onEvent,
+        Action<Exception> onError,
+        Action onCompleted)
+    {
+        var completedGracefully = false;
+        try
+        {
+            using (var call = openCall())
+            {
+                await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token))
+                {
+                    onEvent(evt);
+                }
+            }
+
+            completedGracefully = true;
        }
        catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
        {
            // Normal cancellation — not an error
        }
+        catch (OperationCanceledException) when (cts.IsCancellationRequested)
+        {
+            // Our own Unsubscribe/reconnect cancelled the read — not an error either.
+        }
        catch (Exception ex)
        {
            onError(ex);
@@ -287,6 +340,12 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
            // Remove only our own entry -- a racing reconnect may already own the slot.
            RemoveSubscription(correlationId, cts);
        }
+
+        // Raised outside the try so a throwing callback is not misreported as a stream
+        // fault. Suppressed when our own token was cancelled as the stream ended: that is
+        // a teardown, and the caller has already moved on to a newer stream.
+        if (completedGracefully && !cts.IsCancellationRequested)
+            onCompleted();
    }

    /// <summary>
@@ -400,6 +400,67 @@ public class DebugStreamBridgeActorTests : TestKit
        Assert.Equal("corr-1", factory.ClientFor(GrpcNodeB).SubscribeCalls[0].CorrelationId);
    }

+    // ── WP1.1: graceful (status OK) stream completion is a reconnect trigger ──
+
+    [Fact]
+    public void On_GracefulStreamCompletion_Reopens_On_The_Same_Node()
+    {
+        // The site caps every stream at GrpcMaxStreamLifetime and then ends the RPC with
+        // OK. Pre-fix the client's read loop just finished, the actor was told nothing, and
+        // the debug session went silently deaf for the rest of its life.
+        var (_, factory) = CreateBridgeWithTrackingFactory();
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).SubscribeCalls.Count == 1,
+            TimeSpan.FromSeconds(3));
+
+        factory.ClientFor(GrpcNodeA).SubscribeCalls[0].OnCompleted();
+
+        // Reopened on the SAME node — a clean close is not a fault, so there is nothing to
+        // fail over from — and the finished stream is released, not left zombie.
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).SubscribeCalls.Count == 2,
+            TimeSpan.FromSeconds(5));
+        Assert.Empty(factory.ClientFor(GrpcNodeB).SubscribeCalls);
+        Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).UnsubscribedCorrelationIds);
+    }
+
+    [Fact]
+    public void RepeatedGracefulCompletions_DoNotConsume_TheErrorRetryBudget()
+    {
+        // Five completions — two more than MaxRetries. Had completion been routed through
+        // the error path the session would have terminated on the fourth.
+        var ctx = CreateBridgeActor();
+        ctx.CommProbe.ExpectMsg<SiteEnvelope>();
+        AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3));
+
+        for (var i = 1; i <= 5; i++)
+        {
+            ctx.MockGrpcClient.SubscribeCalls[i - 1].OnCompleted();
+            var expected = i + 1;
+            AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == expected,
+                TimeSpan.FromSeconds(5));
+        }
+
+        Assert.False(ctx.TerminatedFlag[0]);
+    }
+
+    [Fact]
+    public void LateCompletionFromAPreviousStreamGeneration_IsIgnored()
+    {
+        var (_, factory) = CreateBridgeWithTrackingFactory();
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).SubscribeCalls.Count == 1,
+            TimeSpan.FromSeconds(3));
+
+        var firstSub = factory.ClientFor(GrpcNodeA).SubscribeCalls[0];
+        firstSub.OnError(new Exception("NodeA down"));   // gen 1 dies → gen 2 opens on NodeB
+        AwaitCondition(() => factory.ClientFor(GrpcNodeB).SubscribeCalls.Count == 1,
+            TimeSpan.FromSeconds(5));
+
+        // A completion racing out of the dead gen-1 stream must not tear down the live one.
+        firstSub.OnCompleted();
+        Thread.Sleep(300);
+        Assert.DoesNotContain("corr-1", factory.ClientFor(GrpcNodeB).UnsubscribedCorrelationIds);
+        Assert.Single(factory.ClientFor(GrpcNodeB).SubscribeCalls);
+    }
+
    // ── Task 6 (arch review 02, High): teardown/failover unsubscribe is endpoint-safe ──
    // Both paths use TryGet, never GetOrCreate, so cleanup can never open a fresh
    // channel or (with (site,endpoint) keying) touch another session's channel.
@@ -894,9 +955,11 @@ internal class MockSiteStreamGrpcClient : SiteStreamGrpcClient
        string instanceUniqueName,
        Action<object> onEvent,
        Action<Exception> onError,
+        Action onCompleted,
        CancellationToken ct)
    {
-        var subscription = new MockSubscription(correlationId, instanceUniqueName, onEvent, onError, ct);
+        var subscription = new MockSubscription(
+            correlationId, instanceUniqueName, onEvent, onError, onCompleted, ct);
        lock (_lock) { _subscribeCalls.Add(subscription); }

        // Return a task that completes when cancelled (simulates long-running stream)
@@ -916,6 +979,7 @@ internal record MockSubscription(
    string InstanceUniqueName,
    Action<object> OnEvent,
    Action<Exception> OnError,
+    Action OnCompleted,
    CancellationToken CancellationToken);

 /// <summary>
@@ -93,7 +93,8 @@ public class SiteAlarmAggregatorActorTests : TestKit
    }

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

    private sealed class MockSiteAlarmStreamClient : SiteStreamGrpcClient
    {
@@ -107,9 +108,10 @@ public class SiteAlarmAggregatorActorTests : TestKit
        public MockSiteAlarmStreamClient() : base() { }

        public override Task SubscribeSiteAsync(
-            string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> onError, CancellationToken ct)
+            string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> onError,
+            Action onCompleted, CancellationToken ct)
        {
-            lock (_lock) { _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, ct)); }
+            lock (_lock) { _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, onCompleted, ct)); }
            var tcs = new TaskCompletionSource();
            ct.Register(() => tcs.TrySetResult());
            return tcs.Task; // never completes until cancelled (simulates a live stream)
@@ -445,6 +447,82 @@ public class SiteAlarmAggregatorActorTests : TestKit
        Assert.Equal(2, TotalSubs());
    }

+    // ── WP1.1: graceful (status OK) stream completion is a reconnect trigger ──
+
+    [Fact]
+    public void GracefulStreamCompletion_ReopensOnReconcileTick_OnTheSameNode()
+    {
+        // The site ends every stream with OK at its 4h max lifetime. Pre-fix the client's
+        // read loop just finished, nothing was told to the actor, and the site's alarm feed
+        // stayed silently dead until the central node restarted.
+        var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(300));
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
+        seed.CompleteNext();
+
+        factory.ClientFor(GrpcNodeA).Subs[0].OnCompleted();
+
+        // Reopened by the reconcile tick, on the SAME node — a clean close is not a fault,
+        // so there is nothing to fail over from.
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(3));
+        Assert.Empty(factory.ClientFor(GrpcNodeB).Subs);
+        // The finished stream was released, so the site keeps no zombie relay actor.
+        Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).Unsubscribed);
+    }
+
+    [Fact]
+    public void GracefulStreamCompletion_NeitherSpendsNorRefunds_TheErrorRetryBudget()
+    {
+        // Reconcile is far away; reopens are driven explicitly so the budget can be observed.
+        var (actor, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
+        seed.CompleteNext();
+
+        // Spend the whole error budget (MaxRetries = 3), each error flipping the node.
+        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("1"));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
+        factory.ClientFor(GrpcNodeB).Subs[0].OnError(new Exception("2"));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5));
+        factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3"));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5));
+
+        // A graceful completion reopens without flipping the node (budget not spent) …
+        factory.ClientFor(GrpcNodeB).Subs[1].OnCompleted();
+        actor.Tell(new RunReconcile());
+        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 3, TimeSpan.FromSeconds(5));
+        Assert.Equal(2, factory.ClientFor(GrpcNodeA).Subs.Count);
+
+        int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count;
+        var before = TotalSubs();
+
+        // … and without refunding it either: the next error is the 4th, so it exceeds
+        // MaxRetries and the stream is given up rather than reconnected.
+        factory.ClientFor(GrpcNodeB).Subs[2].OnError(new Exception("4"));
+        Thread.Sleep(400);
+        Assert.Equal(before, TotalSubs());
+    }
+
+    [Fact]
+    public void LateCompletionFromAPreviousStreamGeneration_IsIgnored()
+    {
+        var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
+        seed.CompleteNext();
+
+        var firstSub = factory.ClientFor(GrpcNodeA).Subs.Single();
+        firstSub.OnError(new Exception("real fault"));  // gen 1 dies → gen 2 opens on NodeB
+        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
+
+        // A completion racing out of the dead gen-1 stream must not tear down the live
+        // gen-2 stream (which would go deltaless until the next reconcile tick).
+        firstSub.OnCompleted();
+        Thread.Sleep(300);
+        Assert.DoesNotContain("corr-1", factory.ClientFor(GrpcNodeB).Unsubscribed);
+        Assert.Single(factory.ClientFor(GrpcNodeB).Subs);
+    }
+
    // ── R2 T10: live-delta publish coalescing (N6) ──

    [Fact]
@@ -1,4 +1,5 @@
 using Google.Protobuf.WellKnownTypes;
+using Grpc.Core;
 using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
 using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
@@ -277,7 +278,120 @@ public class SiteStreamGrpcClientTests
        var client = SiteStreamGrpcClient.CreateForTesting();

        await Assert.ThrowsAsync<InvalidOperationException>(() =>
-            client.SubscribeSiteAsync("corr", _ => { }, _ => { }, CancellationToken.None));
+            client.SubscribeSiteAsync("corr", _ => { }, _ => { }, () => { }, CancellationToken.None));
+    }
+
+    // --- WP1.1: graceful (status OK) stream completion is reported, not swallowed ---
+
+    [Fact]
+    public async Task ConsumeStream_ServerEndsStreamWithOk_InvokesOnCompleted_NotOnError()
+    {
+        // The site server caps every stream at GrpcMaxStreamLifetime (4h) and then ends the
+        // RPC with OK. That surfaces here as a read loop that simply runs out of events.
+        var client = SiteStreamGrpcClient.CreateForTesting();
+        var cts = new CancellationTokenSource();
+        var events = new List<SiteStreamEvent>();
+        Exception? error = null;
+        var completed = 0;
+
+        await client.ConsumeStreamAsync(
+            "corr-ok",
+            cts,
+            () => FakeCall(new StubStreamReader(new SiteStreamEvent { CorrelationId = "corr-ok" })),
+            events.Add,
+            ex => error = ex,
+            () => completed++);
+
+        Assert.Single(events);
+        Assert.Null(error);
+        Assert.Equal(1, completed);
+    }
+
+    [Fact]
+    public async Task ConsumeStream_StreamFaults_InvokesOnError_NotOnCompleted()
+    {
+        var client = SiteStreamGrpcClient.CreateForTesting();
+        var cts = new CancellationTokenSource();
+        Exception? error = null;
+        var completed = 0;
+
+        await client.ConsumeStreamAsync(
+            "corr-fault",
+            cts,
+            () => FakeCall(new StubStreamReader(
+                new RpcException(new Status(StatusCode.Unavailable, "site gone")))),
+            _ => { },
+            ex => error = ex,
+            () => completed++);
+
+        Assert.IsType<RpcException>(error);
+        Assert.Equal(0, completed);
+    }
+
+    [Fact]
+    public async Task ConsumeStream_OwnCancellation_InvokesNeitherCallback()
+    {
+        // Our own Unsubscribe/reconnect is a teardown, not a fault and not a graceful end:
+        // the caller has already moved on to a newer stream.
+        var client = SiteStreamGrpcClient.CreateForTesting();
+        var cts = new CancellationTokenSource();
+        await cts.CancelAsync();
+        Exception? error = null;
+        var completed = 0;
+
+        await client.ConsumeStreamAsync(
+            "corr-cancel",
+            cts,
+            () => FakeCall(new StubStreamReader()),
+            _ => { },
+            ex => error = ex,
+            () => completed++);
+
+        Assert.Null(error);
+        Assert.Equal(0, completed);
+    }
+
+    private static AsyncServerStreamingCall<SiteStreamEvent> FakeCall(StubStreamReader reader) =>
+        new(reader,
+            Task.FromResult(new Metadata()),
+            () => Status.DefaultSuccess,
+            () => new Metadata(),
+            () => { });
+
+    /// <summary>
+    /// Server stream stand-in: yields the queued events, then either ends the stream (the
+    /// status-OK completion the site's max-lifetime cap produces) or throws.
+    /// </summary>
+    private sealed class StubStreamReader : IAsyncStreamReader<SiteStreamEvent>
+    {
+        private readonly Queue<SiteStreamEvent> _events;
+        private readonly Exception? _fault;
+
+        public StubStreamReader(params SiteStreamEvent[] events)
+        {
+            _events = new Queue<SiteStreamEvent>(events);
+        }
+
+        public StubStreamReader(Exception fault)
+            : this()
+        {
+            _fault = fault;
+        }
+
+        public SiteStreamEvent Current { get; private set; } = null!;
+
+        public Task<bool> MoveNext(CancellationToken cancellationToken)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            if (_events.Count > 0)
+            {
+                Current = _events.Dequeue();
+                return Task.FromResult(true);
+            }
+
+            return _fault is null ? Task.FromResult(false) : Task.FromException<bool>(_fault);
+        }
    }

    // --- Communication-003 regression tests ---
@@ -29,7 +29,7 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
        public HangingClient() : base() { }
        public override Task SubscribeSiteAsync(
            string correlationId, Action<Commons.Messages.Streaming.AlarmStateChanged> onAlarmEvent,
-            Action<Exception> onError, CancellationToken ct)
+            Action<Exception> onError, Action onCompleted, CancellationToken ct)
        {
            var tcs = new TaskCompletionSource();
            ct.Register(() => tcs.TrySetResult());
@@ -279,7 +279,8 @@ public class SiteAlarmStreamEndToEndTests : TestKit
    private sealed class NoopSiteStreamClient : SiteStreamGrpcClient
    {
        public override Task SubscribeSiteAsync(
-            string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> onError, CancellationToken ct)
+            string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> onError,
+            Action onCompleted, CancellationToken ct)
        {
            var tcs = new TaskCompletionSource();
            ct.Register(() => tcs.TrySetResult());