fix(galaxy): #865 review round 1 - stale-session pump, data-path stream wait

- EnsureEventPumpStarted replaces a pump created for an older registry
  session (the reopen -> replay window still held the dead session's pump,
  so an advise there lost its initial value and the replay skipped it);
  RestartEventPumpForReplay keeps a pump already on the current session.
- SubscribeAsync, read-via-subscribe and ReplayAsync also await the
  bounded StreamRequested wait before advising, not just the probes.
- TryEnsureEventPumpStarted tolerates a teardown flipping _runtimeStopping
  mid-check; the probe hook honours an already-cancelled seed token.
- StreamRequested is documented as narrowing, not closing, the attach race.
- Tests: reopen-window probe test, EventPump.StreamRequested tests, and the
  FixRound1 fresh-pump assertion now waits on a positive control.

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs

@@ -1053,6 +1053,10 @@ public sealed class GalaxyDriver
         // subscriptions to replay the dead pump used to stay in _eventPump, so every later
         // Subscribe/Read reused it and was silently deaf while health stayed Healthy.
         RestartEventPumpForReplay();
+        // #865: every re-advise below pushes its initial value once, to whatever stream is attached.
+        EventPump? replayPump;
+        lock (_pumpLock) replayPump = _eventPump;
+        if (replayPump is not null) await AwaitStreamRequestedAsync(replayPump, cancellationToken).ConfigureAwait(false);
 
         // Only subscriptions still holding the previous session's handles: one registered after the
         // reopen is already live on the recreated session (re-subscribing it would double-advise).
@@ -1308,6 +1312,10 @@ public sealed class GalaxyDriver
         {
             old = _eventPump;
             if (old is null) return; // pump never started — nothing to restart
+            // #865: a pump created after the reopen's fence (an advise in the reopen → replay window
+            // started it via EnsureEventPumpStarted) already streams the recreated session, and advises
+            // registered under it may be waiting for their initial values on it — keep it.
+            if (old.Session == _subscriptions.CurrentSession) return;
             _eventPump = null;
         }
 
@@ -1681,7 +1689,10 @@ public sealed class GalaxyDriver
         var readSubscriber = _subscriber!;
         // A live pump must exist (values arrive through it); the read listens on ReadDataChange, not
         // on this instance, so it survives a reconnect's pump restart.
-        EnsureEventPumpStarted();
+        var readPump = EnsureEventPumpStarted();
+        // #865: a read waits ONLY for the initial value, which reaches only an already-requested stream.
+        // Not the caller's token: a cancel here must still yield per-ref Bad snapshots, and the wait is bounded.
+        await AwaitStreamRequestedAsync(readPump, CancellationToken.None).ConfigureAwait(false);
         var subscriptionId = _subscriptions.NextSubscriptionId();
 
         // Pre-allocate one TaskCompletionSource per full-reference so the OnDataChange
@@ -1927,6 +1938,8 @@ public sealed class GalaxyDriver
         var generation = Volatile.Read(ref _runtimeGeneration);
         var subscriber = _subscriber;
         var pump = EnsureEventPumpStarted();
+        // #865: the advise's initial value reaches only an already-requested stream (no replay).
+        await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false);
         var subscriptionId = _subscriptions.NextSubscriptionId();
 
         if (fullReferences.Count == 0)
@@ -2081,8 +2094,18 @@ public sealed class GalaxyDriver
     ///     <see cref="SubscriptionRegistry"/> reverse map, not by spinning a pump per
     ///     subscription.
     /// </summary>
+    /// <remarks>
+    ///     #865: a pump created for an OLDER registry session is replaced, not reused. Between
+    ///     <see cref="ReopenAsync"/>'s session swap and <see cref="ReplayAsync"/>'s pump restart the
+    ///     current pump still streams the dead gateway session; an advise made in that window goes to the
+    ///     recreated session, so its initial value would reach no stream — and, registered under the
+    ///     current session, the replay would not re-advise it either. A pump whose session is current
+    ///     was created after the reopen's second fence, so its stream is on the recreated session.
+    /// </remarks>
     private EventPump EnsureEventPumpStarted()
     {
+        EventPump? stale = null;
+        EventPump pump;
         lock (_pumpLock)
         {
             // Checked BEFORE the reuse: the flag is set before the teardown takes the pump under this
@@ -2094,7 +2117,12 @@ public sealed class GalaxyDriver
                 throw new InvalidOperationException(
                     "GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync.");
             }
-            if (_eventPump is not null) return _eventPump;
+            if (_eventPump is not null && _eventPump.Session == _subscriptions.CurrentSession) return _eventPump;
+            if (_eventPump is not null)
+            {
+                stale = _eventPump;
+                stale.OnDataChange -= OnPumpDataChange;
+            }
             _eventPump = new EventPump(
                 _subscriber!, _subscriptions, _logger,
                 channelCapacity: _options.MxAccess.EventPumpChannelCapacity,
@@ -2103,13 +2131,26 @@ public sealed class GalaxyDriver
                 onReplayGap: OnEventPumpReplayGap);
             _eventPump.OnDataChange += OnPumpDataChange;
             _eventPump.Start();
-            return _eventPump;
+            pump = _eventPump;
         }
+
+        if (stale is not null) DisposeStalePumpInBackground(stale);
+        return pump;
     }
 
+    /// <summary>Dispose a replaced pump off the caller's path (its stream is dead; the dispose only joins its loops).</summary>
+    private void DisposeStalePumpInBackground(EventPump stale) => _ = Task.Run(async () =>
+    {
+        try { await stale.DisposeAsync().ConfigureAwait(false); }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "GalaxyDriver {InstanceId} stale-session EventPump dispose failed.", _driverInstanceId);
+        }
+    });
+
     /// <summary>
     ///     <see cref="EnsureEventPumpStarted"/> for infrastructure callers (the probe watcher) that
-    ///     must not fail because the runtime is going away: false — and no pump — while a teardown is in
+    ///     must not fail because the runtime is going away: null — and no pump — while a teardown is in
     ///     progress or there is no subscriber.
     /// </summary>
     /// <returns>The running (or just started) pump; null when none may be started.</returns>
@@ -2118,36 +2159,58 @@ public sealed class GalaxyDriver
         lock (_pumpLock)
         {
             if (_runtimeStopping || _disposed || _subscriber is null) return null;
-            return EnsureEventPumpStarted(); // re-enters _pumpLock (Lock is reentrant); cannot throw here
+            try
+            {
+                return EnsureEventPumpStarted(); // re-enters _pumpLock (Lock is reentrant)
+            }
+            catch (InvalidOperationException)
+            {
+                // _runtimeStopping is written without _pumpLock; a teardown that flipped it between the
+                // check above and EnsureEventPumpStarted's own check is the same "no pump" answer.
+                return null;
+            }
         }
     }
 
-    /// <summary>Upper bound on how long a probe subscribe waits for the pump's StreamEvents call to be issued.</summary>
-    internal TimeSpan ProbeStreamRequestWait { get; set; } = TimeSpan.FromSeconds(5);
+    /// <summary>Upper bound on how long an advise waits for the pump's StreamEvents call to be started.</summary>
+    internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(5);
 
     /// <summary>
-    ///     #865 — the probe watcher's pre-subscribe hook: start the pump (if this runtime may) and wait,
-    ///     bounded, until its StreamEvents call has been issued, so the gateway has a subscriber to push
-    ///     the probe's initial ScanState value to. A fresh stream (after_worker_sequence = 0) is never
-    ///     replayed, so an initial value pushed before it attaches is lost — and ScanState rarely changes
-    ///     again. On timeout the subscribe proceeds anyway (logged): a late stream is no worse than before.
+    ///     #865 — wait, bounded, until <paramref name="pump"/> has started its StreamEvents call, so an
+    ///     advise made next has a stream for its initial value. A fresh stream (after_worker_sequence = 0)
+    ///     is never replayed, so an initial value pushed before it attaches is lost — and a slow-changing
+    ///     attribute (ScanState, a setpoint) may not publish again for a long time. On timeout the caller
+    ///     proceeds (logged): a late stream is no worse than the pre-#865 behaviour.
     /// </summary>
-    private async Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken)
+    private async Task AwaitStreamRequestedAsync(EventPump pump, CancellationToken cancellationToken)
     {
-        var pump = TryEnsureEventPumpStarted();
-        if (pump is null) return;
         try
         {
-            await pump.StreamRequested.WaitAsync(ProbeStreamRequestWait, cancellationToken).ConfigureAwait(false);
+            await pump.StreamRequested.WaitAsync(StreamRequestWait, cancellationToken).ConfigureAwait(false);
         }
         catch (TimeoutException)
         {
             _logger.LogWarning(
-                "GalaxyDriver {InstanceId} event stream was not requested within {Wait} — subscribing the platform probes anyway; their initial ScanState may be missed.",
-                _driverInstanceId, ProbeStreamRequestWait);
+                "GalaxyDriver {InstanceId} event stream was not started within {Wait} — advising anyway; initial values may be missed.",
+                _driverInstanceId, StreamRequestWait);
         }
     }
 
+    /// <summary>
+    ///     #865 — the probe watcher's pre-subscribe hook: start the pump (if this runtime may) and wait for
+    ///     its stream (<see cref="AwaitStreamRequestedAsync"/>). A probe's initial ScanState value is usually
+    ///     its only event, and a driver with no authored tags never Reads/Subscribes, so without this its
+    ///     pump — and every host status — never existed.
+    /// </summary>
+    private async Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken)
+    {
+        // A seed pass a teardown already cancelled must not start a pump after the teardown finished.
+        cancellationToken.ThrowIfCancellationRequested();
+        var pump = TryEnsureEventPumpStarted();
+        if (pump is null) return;
+        await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false);
+    }
+
     /// <summary>
     ///     Replay-gap callback for the <see cref="EventPump"/>. The gateway told us it evicted
     ///     events we can never receive, so every cached value may be stale by an unknown amount.

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs

@@ -148,15 +148,27 @@ internal sealed class EventPump : IAsyncDisposable
 
     private readonly int _session;
 
+    /// <summary>
+    ///     The registry session this pump was created for (#865). A pump of an older session was opened
+    ///     against a gateway session a reconnect replaced; its stream will never carry the new session's events.
+    /// </summary>
+    internal int Session => _session;
+
     private readonly TaskCompletionSource _streamRequested = new(TaskCreationOptions.RunContinuationsAsynchronously);
 
     /// <summary>
-    ///     Completes once the stream loop has issued its <c>StreamEvents</c> call (the first
-    ///     <c>MoveNextAsync</c> runs the subscriber's iterator synchronously up to the RPC) — or once the
-    ///     loop has ended without getting that far, so a waiter never hangs on a dead pump. #865: the
-    ///     gateway does not replay a fresh stream, so a caller whose only expected event is the initial
-    ///     value of an advise it is about to make waits on this first.
+    ///     Completes once the stream loop has STARTED its <c>StreamEvents</c> call (the first
+    ///     <c>MoveNextAsync</c> runs the subscriber's iterator synchronously up to the gRPC call creation) —
+    ///     or once the loop has ended without getting that far, so a waiter never hangs on a dead pump.
+    ///     #865: the gateway does not replay a fresh stream, so a caller whose only expected event is the
+    ///     initial value of an advise it is about to make waits on this first.
     /// </summary>
+    /// <remarks>
+    ///     ⚠️ This NARROWS the stream-attach race, it does not close it: the HTTP/2 request is sent
+    ///     asynchronously and the gateway attaches the subscriber after auth, concurrently with any other
+    ///     RPC. In practice the attach wins (an advise's initial value needs a worker IPC round trip plus an
+    ///     MXAccess advise), but the gateway offers no attach acknowledgement to await.
+    /// </remarks>
     public Task StreamRequested => _streamRequested.Task;
 
     /// <summary>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamRequestedTests.cs

@@ -0,0 +1,62 @@
+using System.Runtime.CompilerServices;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
+
+/// <summary>
+///     #865: <see cref="EventPump.StreamRequested"/> — what advise paths wait on so an initial value is
+///     never pushed before the stream is requested. It must complete once the RPC is started (before any
+///     event), and must never strand a waiter on a pump whose loop died before getting that far.
+/// </summary>
+public sealed class EventPumpStreamRequestedTests
+{
+    private sealed class ScriptedSubscriber(Func<CancellationToken, IAsyncEnumerable<MxEvent>> stream) : IGalaxySubscriber
+    {
+        public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
+            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
+            => Task.FromResult<IReadOnlyList<SubscribeResult>>([]);
+
+        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;
+
+        public IAsyncEnumerable<MxEvent> StreamEventsAsync(CancellationToken cancellationToken) => stream(cancellationToken);
+    }
+
+    /// <summary>A stream that is requested and then never yields: StreamRequested completes; the RPC started first.</summary>
+    [Fact]
+    public async Task Completes_once_the_stream_is_requested_before_any_event()
+    {
+        var rpcStarted = 0;
+        async IAsyncEnumerable<MxEvent> Never([EnumeratorCancellation] CancellationToken ct)
+        {
+            Interlocked.Exchange(ref rpcStarted, 1); // the synchronous first step — where a real call is created
+            await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
+            yield break;
+        }
+
+        await using var pump = new EventPump(new ScriptedSubscriber(Never), new SubscriptionRegistry());
+        pump.StreamRequested.IsCompleted.ShouldBeFalse("not before Start");
+        pump.Start();
+
+        await TestWait.ForAsync(pump.StreamRequested, "StreamRequested must complete once the stream call is started");
+        Volatile.Read(ref rpcStarted).ShouldBe(1, "StreamRequested must not precede the stream call");
+    }
+
+    /// <summary>A subscriber that throws before reaching the RPC still completes StreamRequested (no stranded waiter).</summary>
+    [Fact]
+    public async Task Completes_when_the_stream_fails_before_the_rpc()
+    {
+        var faulted = new TaskCompletionSource<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
+        await using var pump = new EventPump(
+            new ScriptedSubscriber(_ => throw new InvalidOperationException("not connected")),
+            new SubscriptionRegistry(),
+            onStreamFault: ex => faulted.TrySetResult(ex));
+        pump.Start();
+
+        await TestWait.ForAsync(pump.StreamRequested, "a dead pump must not strand a StreamRequested waiter");
+        (await TestWait.ForAsync(faulted.Task, "positive control: the failure is still reported as a stream fault"))
+            .ShouldBeOfType<InvalidOperationException>();
+    }
+}

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs

@@ -333,7 +333,10 @@ public sealed class GalaxyDriverFixRound1Tests
         await driver.InitializeAsync("{}", CancellationToken.None);
         // #865: the new runtime's background probe seed starts a pump of its own (its advises need a live
         // stream) — but it must be a FRESH one, never the dead subscriber's.
-        driver.EventPumpInstanceForTests.ShouldNotBeSameAs(oldPump, "the next runtime must not inherit the old pump");
+        await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1,
+            "positive control: the new runtime's background seed must advise its platform");
+        driver.EventPumpInstanceForTests.ShouldNotBeNull("the seed's probe advise must have started a pump")
+            .ShouldNotBeSameAs(oldPump, "the next runtime must not inherit the old pump");
     }
 
     // ───────────────────────────── IMPORTANT 3: bounded gateway close ─────────────────────────────

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs

@@ -193,6 +193,51 @@ public sealed class ProbeInitialValueTests
         sub.DroppedInitialValues.ShouldBe(0);
     }
 
+    /// <summary>
+    ///     Review finding: a probe advised in the reopen → replay window (the session was swapped, the old
+    ///     pump still streams the dead session) must still get its initial value — the stale-session pump
+    ///     is replaced before the advise, and the replay keeps the replacement (it already streams the
+    ///     recreated session). Reusing the old pump bound the value to the old session, so it never routed,
+    ///     and the replay skipped the platform (its handle is current-session).
+    /// </summary>
+    [Fact]
+    public async Task Probe_advised_between_reopen_and_replay_still_gets_its_initial_value()
+    {
+        var sub = new GatewayLikeSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub,
+            platformSource: new FixedPlatformSource(Platforms))
+        {
+            RecreateOverrideForTests = _ => Task.CompletedTask,
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        await TestWait.UntilAsync(
+            () => Platforms.All(p => StateOf(driver, p) == HostState.Running),
+            "positive control: the initial seed reports Running",
+            () => Describe(driver, sub));
+        var oldPump = driver.EventPumpInstanceForTests;
+
+        await driver.InvokeReopenForTestAsync(CancellationToken.None); // session swapped; no replay yet
+        sub.ScanStateRunning = false;
+        await driver.ProbeWatcherForTests.ShouldNotBeNull()
+            .SyncPlatformsAsync([.. Platforms, "LatePlatform"], CancellationToken.None);
+
+        await TestWait.UntilAsync(
+            () => StateOf(driver, "LatePlatform") == HostState.Stopped,
+            "a probe advised in the reopen window must get its initial value",
+            () => Describe(driver, sub));
+        var windowPump = driver.EventPumpInstanceForTests;
+        windowPump.ShouldNotBeSameAs(oldPump, "the stale-session pump must have been replaced before the advise");
+
+        await driver.InvokeReplayForTestAsync(CancellationToken.None);
+        driver.EventPumpInstanceForTests.ShouldBeSameAs(windowPump, "the replay keeps a pump already on the current session");
+        await TestWait.UntilAsync(
+            () => Platforms.All(p => StateOf(driver, p) == HostState.Stopped),
+            "the replay re-advises the stale probes and their initial values arrive on the kept pump",
+            () => Describe(driver, sub));
+        StateOf(driver, "LatePlatform").ShouldBe(HostState.Stopped);
+        sub.DroppedInitialValues.ShouldBe(0);
+    }
+
     /// <summary>
     ///     A probe sync on a driver whose pump was never started (no platform source, e.g. the
     ///     DiscoverAsync path) still starts it before advising — the watcher's pre-subscribe hook, not just