fix(centralui): stop debug-stream sessions abandoned mid-start — dispose-vs-connect race (arch-review 2026-08-19 P6-6)

A debug-stream session created after the consumer had already gone away was
leaked permanently. Both consumers awaited StartStreamAsync (up to 30s) and only
then published the session; a circuit teardown / client disconnect inside that
window ran the release path first, found nothing registered and stopped nothing.
The session stayed registered in the singleton DebugStreamService, whose
keepalive renews registered sessions forever (the 5-minute orphan reaper only
collects UNregistered ones), pinning the site gRPC subscription, the
DebugStreamBridgeActor and the dead consumer's object graph until process
restart.

Both call sites now publish the session and re-check liveness under a single
monitor, so exactly one side tears it down and neither can miss it:

- DebugView.Connect/Dispose/Disconnect share a _sessionLock; Dispose sets
  _disposed and takes the session in one step, Connect publishes only if not
  disposed and otherwise stops the session it just created. Dispose can run off
  the renderer's thread, so the Blazor sync context is not relied on. Disconnect
  and the onTerminated callback go through the same take-once helper, which also
  makes a repeated Dispose stop-once instead of stop-again.
- DebugStreamHub uses the per-connection Items monitor: OnDisconnectedAsync marks
  the connection gone before releasing, SubscribeInstance publishes only if
  neither that flag nor ConnectionAborted is set, and UnsubscribeInstance takes
  the id atomically.

DebugStreamService.StartStreamAsync/StopStream become virtual purely as a test
seam (no behaviour change, keepalive/orphan contract untouched) so the race can
be driven deterministically by a gated stub.

Tests: DebugViewConnectDisposeRaceTests (3) + DebugStreamHubSubscribeRaceTests
(3). Verified fail-unfixed: with the two source fixes stashed, both
dispose-mid-start tests report zero StopStream calls and the DebugView
double-dispose test reports two.

src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Deployment/DebugView.razor

@@ -343,6 +343,16 @@
     }
 
     private DebugStreamSession? _session;
+
+    /// <summary>
+    /// Guards the <see cref="_session"/> slot together with <see cref="_disposed"/>
+    /// (arch-review 2026-08-19 P6-6). <c>Connect</c> awaits <c>StartStreamAsync</c> for up
+    /// to 30s; the circuit can be torn down inside that window, and <c>Dispose</c> may run
+    /// on a non-UI thread, so publishing the session and observing disposal must be one
+    /// atomic step. Whoever loses the race is the one that stops the stream.
+    /// </summary>
+    private readonly object _sessionLock = new();
+
     private ToastNotification _toast = default!;
 
     private string? _initError;
@@ -468,7 +478,7 @@
                 onTerminated: () =>
                 {
                     _connected = false;
-                    _session = null;
+                    _ = TakeSession();
                     // CentralUI-009: skip the toast/render if already disposed.
                     if (_disposed) return;
                     _ = SafeInvokeAsync(() =>
@@ -490,7 +500,25 @@
                 return;
             }
 
-            _session = session;
+            // P6-6: the component may have been disposed while StartStreamAsync was in
+            // flight. Dispose() would have found _session still null and stopped nothing,
+            // so this session would stay registered in the singleton DebugStreamService —
+            // whose keepalive renews registered sessions forever, pinning the site
+            // subscription, the bridge actor and this dead circuit until process restart.
+            // Publish and re-check disposal atomically so exactly one side tears it down.
+            bool abandoned;
+            lock (_sessionLock)
+            {
+                abandoned = _disposed;
+                if (!abandoned) _session = session;
+            }
+
+            if (abandoned)
+            {
+                DebugStreamService.StopStream(session.SessionId);
+                _connecting = false;
+                return;
+            }
 
             // Populate initial state from snapshot
             _attributeValues.Clear();
@@ -521,12 +549,26 @@
         _connecting = false;
     }
 
+    /// <summary>
+    /// Atomically clears the <see cref="_session"/> slot and returns what was there, so
+    /// only the caller that actually took the session stops it (P6-6).
+    /// </summary>
+    private DebugStreamSession? TakeSession()
+    {
+        lock (_sessionLock)
+        {
+            var session = _session;
+            _session = null;
+            return session;
+        }
+    }
+
     private async Task Disconnect()
     {
-        if (_session != null)
+        var session = TakeSession();
+        if (session != null)
         {
-            DebugStreamService.StopStream(_session.SessionId);
-            _session = null;
+            DebugStreamService.StopStream(session.SessionId);
         }
 
         // Clear persisted selection — user explicitly disconnected
@@ -725,10 +767,20 @@
     {
         // CentralUI-009: mark disposed first so any in-flight stream callback
         // sees the flag and no-ops, then stop the stream synchronously.
-        _disposed = true;
-        if (_session != null)
+        // P6-6: the flag is set and the session taken under the same lock Connect
+        // publishes through, so a Connect still awaiting StartStreamAsync either
+        // hands the session over here or stops it itself — never neither.
+        DebugStreamSession? session;
+        lock (_sessionLock)
         {
-            DebugStreamService.StopStream(_session.SessionId);
+            _disposed = true;
+            session = _session;
+            _session = null;
+        }
+
+        if (session != null)
+        {
+            DebugStreamService.StopStream(session.SessionId);
         }
     }
 }

src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs

@@ -81,7 +81,12 @@ public class DebugStreamService : IDisposable
     /// <param name="onTerminated">Callback invoked when the stream terminates.</param>
     /// <param name="ct">A cancellation token that can be used to cancel the operation.</param>
     /// <returns>A debug stream session with the initial snapshot.</returns>
-    public async Task<DebugStreamSession> StartStreamAsync(
+    /// <remarks>
+    /// Virtual purely as a test seam: the consumers' dispose-vs-start race (P6-6) can only
+    /// be driven deterministically by a stub that holds the start open. Behaviour and the
+    /// keepalive/orphan contract are unchanged.
+    /// </remarks>
+    public virtual async Task<DebugStreamSession> StartStreamAsync(
         int instanceId,
         Action<object> onEvent,
         Action onTerminated,
@@ -209,7 +214,8 @@ public class DebugStreamService : IDisposable
     /// Stops an active debug stream session.
     /// </summary>
     /// <param name="sessionId">The session ID of the debug stream to stop.</param>
-    public void StopStream(string sessionId)
+    /// <remarks>Virtual purely as a test seam — see <see cref="StartStreamAsync"/>.</remarks>
+    public virtual void StopStream(string sessionId)
     {
         if (_sessions.TryRemove(sessionId, out var bridgeActor))
         {

src/ZB.MOM.WW.ScadaBridge.ManagementService/DebugStreamHub.cs

@@ -19,6 +19,7 @@ public class DebugStreamHub : Hub
     private const string SessionIdKey = "DebugStreamSessionId";
     private const string RolesKey = "DebugStreamRoles";
     private const string PermittedSiteIdsKey = "DebugStreamPermittedSiteIds";
+    private const string DisconnectedKey = "DebugStreamDisconnected";
 
     /// <summary>
     /// Pure site-scope authorization check for a debug-stream subscription.
@@ -192,7 +193,35 @@ public class DebugStreamHub : Hub
                             CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);
                 });
 
-            Context.Items[SessionIdKey] = session.SessionId;
+            // arch-review 2026-08-19 P6-6: the client may have disconnected while
+            // StartStreamAsync was in flight (up to 30s). OnDisconnectedAsync already ran
+            // and found no session id to release, so publishing it here would leave the
+            // session registered in the singleton DebugStreamService — whose keepalive
+            // renews registered sessions forever, pinning the site subscription and its
+            // bridge actor until process restart. Publish and re-check liveness under the
+            // per-connection Items monitor (the same one UnsubscribeInstance takes) so
+            // exactly one side tears the session down.
+            var abandoned = false;
+            lock (Context.Items)
+            {
+                if (Context.Items.ContainsKey(DisconnectedKey) || Context.ConnectionAborted.IsCancellationRequested)
+                {
+                    abandoned = true;
+                }
+                else
+                {
+                    Context.Items[SessionIdKey] = session.SessionId;
+                }
+            }
+
+            if (abandoned)
+            {
+                _debugStreamService.StopStream(session.SessionId);
+                _logger.LogInformation(
+                    "DebugStreamHub: {ConnectionId} disconnected while subscribing to instance {InstanceId} — session released",
+                    connectionId, instanceId);
+                return;
+            }
 
             // Send the initial snapshot
             await Clients.Caller.SendAsync("OnSnapshot", session.InitialSnapshot);
@@ -214,18 +243,45 @@ public class DebugStreamHub : Hub
     /// <returns>A task that represents the asynchronous operation.</returns>
     public Task UnsubscribeInstance()
     {
-        if (Context.Items.TryGetValue(SessionIdKey, out var sessionIdObj) && sessionIdObj is string sessionId)
+        var sessionId = TakeSessionId();
+        if (sessionId is not null)
         {
             _debugStreamService.StopStream(sessionId);
-            Context.Items.Remove(SessionIdKey);
             _logger.LogInformation("DebugStreamHub: {ConnectionId} unsubscribed", Context.ConnectionId);
         }
         return Task.CompletedTask;
     }
 
+    /// <summary>
+    /// Atomically removes and returns the session id registered for this connection, so a
+    /// release path and the late publish in <see cref="SubscribeInstance"/> cannot both
+    /// claim (or both miss) the same session — see P6-6 there.
+    /// </summary>
+    /// <returns>The session id that was registered, or <see langword="null"/> if none was.</returns>
+    private string? TakeSessionId()
+    {
+        lock (Context.Items)
+        {
+            if (Context.Items.TryGetValue(SessionIdKey, out var sessionIdObj) && sessionIdObj is string sessionId)
+            {
+                Context.Items.Remove(SessionIdKey);
+                return sessionId;
+            }
+        }
+        return null;
+    }
+
     /// <inheritdoc />
     public override async Task OnDisconnectedAsync(Exception? exception)
     {
+        // P6-6: mark the connection gone BEFORE releasing, so a SubscribeInstance still
+        // awaiting StartStreamAsync sees the flag and stops its own session rather than
+        // registering it against a connection nothing will ever unsubscribe.
+        lock (Context.Items)
+        {
+            Context.Items[DisconnectedKey] = true;
+        }
+
         await UnsubscribeInstance();
         await base.OnDisconnectedAsync(exception);
     }