Merge #859/#861 Galaxy: early events parked across the subscribe-reply race; per-session fencing across reconnects

Four independent reviews (blockers fixed each round).

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

docs/drivers/Galaxy.md

@@ -172,6 +172,7 @@ Every (re)open of `StreamAlarms` replays the gateway's active-alarm snapshot. `G
 - **Replay gaps** (`galaxy.events.replay_gaps`): the gateway signals that events were evicted from its replay ring before we could read them. The driver responds by re-issuing `SubscribeBulk` for every tracked subscription — MXAccess delivers an initial value per item on subscribe, so re-subscribing *is* the re-snapshot the contract asks for. It does **not** restart the event pump (the stream carrying the sentinel is healthy) and it honours `Reconnect:ReplayOnSessionLost=false` by logging and standing down, on the grounds that an operator who delegated recovery to the gateway did not ask for a re-subscribe on a different trigger.
   ⚠️ This cannot currently fire: the sentinel is only emitted on a stream opened with `after_worker_sequence > 0`, and the driver opens with `0`, which the gateway treats as "fresh stream, not a resume". Enabling resume is a bigger change than it looks — the sequence is only meaningful **within one gateway session**, and the reconnect path recreates the session, so a remembered sequence replayed against a fresh session would set the server's live watermark *above* every event it is about to send, silencing the stream rather than recovering it.
 - **Undecodable values** (`galaxy.events.undecodable`): a value kind this driver has no CLR representation for is published as **`BadTypeMismatch` with a null value**, and the first occurrence of each distinct kind is logged. It is *not* published as a null carrying the event's own quality: an MXAccess event can be Good-quality while the payload is one we cannot map, and a good null is indistinguishable from a genuine null reading — a client would consume a value it never received as a real one, with the failure visible only in a log. Non-zero here means the gateway is sending a wire type `MxValueDecoder` has no mapping for and those readings are being lost; the fix is a decoder mapping, not a quality change.
+- **Initial values can overtake the subscribe reply (#859)** (`galaxy.events.early_parked` / `early_replayed` / `early_overflow`): mxaccessgw gives no reply-before-event ordering — the worker frames the SubscribeBulk reply on a thread-pool continuation while the STA is already pumping the OnDataChange that Advise queued, and the reply (unary Invoke) and the event (StreamEvents) are separate HTTP/2 streams anyway. An event for an unregistered handle is therefore only "stale after unsubscribe" when no subscribe is in flight. Every SubscribeBulk call site (`SubscribeAsync`, the subscribe-once `ReadAsync` — which waits only for the initial value and would otherwise read `BadTimeout` — the reconnect/replay-gap `ResubscribeAllAsync` (one window per subscription, not across the whole replay), `PerPlatformProbeWatcher`) opens `SubscriptionRegistry.BeginPendingSubscribe()` before the RPC and closes it right after registering, never later; inside the window unknown-handle events are parked, promoted by the `Register`/`Rebind` that binds the handle — map writes and promotion are one critical section, so no later event can resolve in between — and replayed on the pump's dispatch thread before any later event for that handle. Held events (parked + promoted-but-unreplayed) are bounded at 100 000 together. Unclaimed parked events are trimmed by **window sequence**, not only when every window has closed (concurrent reads keep some window open almost permanently): each window gets a monotonically increasing sequence when it opens, each parked event is stamped with the newest sequence, and once every window that opened at or before that stamp has closed the event is discarded — outside any window the stale drop is unchanged. **Sessions (#861):** item handles are per gateway session and the next session reuses the numbers. On reconnect `ReopenAsync` calls `SubscriptionRegistry.StartNewSession()` both before and after the gateway session swap (`ReplayAsync` does it only when no reopen ran), and the replay re-subscribes only subscriptions still holding the previous session's handles (one registered after the swap is already live). The pump is always restarted, even with nothing to replay. `StartNewSession()` discards everything held, **invalidates the reverse map** (subscriptions are kept; new-session events park until the replay's per-subscription `Rebind` promotes them, instead of resolving to the tag that owned the number before), and bumps a session number. Registration and rebind are refused for a stale session, so a `SubscribeAsync` whose reply straddled the reconnect is **re-issued on the new session** (up to 3 attempts), a subscribe-once read returns `BadCommunicationError`, and dead old-session handle numbers are never sent to `UnsubscribeBulk` (a read rebound by the replay releases the handles the registry holds, not its original ones); an old session's pump can no longer resolve through the map. Reads listen on the driver, not on a pump instance, so a read survives the reconnect's pump restart. `PerPlatformProbeWatcher` records the session with each probe handle: only current-session handles are published (fenced `TryUpsert`), unsubscribed or purged, and a seed reply that straddled a reconnect is re-issued. Purges are fenced by the session captured before the unsubscribe RPC. **Interim (I-3):** a completed `UnsubscribeBulk` purges anything parked for its handles, so a trailing event of a removed item is not promoted onto a reused handle number within one session. It cannot catch an event still in flight after the unsubscribe, and it also drops a genuine early value for a number reused that fast (a timeout rather than another tag's value). The complete fix needs a worker sequence on the SubscribeBulk reply so the driver can tell pre-subscribe events from post-subscribe ones — an optional cross-repo change in mxaccessgw.
 - **Duplicate `attributeRef` (#605 / 06-S-13)**: two authored RawPaths dialling the same Galaxy attribute are both dialled, but the reverse map is first-wins, so value fan-out and alarm `ConditionId` surface under the first RawPath only. The loser is logged at Warning and recorded in the skipped-tag ledger with the winner path in the reason.
 - **Subscribe-once reads are bounded (#605 / 06-S-3)**: `ReadAsync` on the production path (SubscribeBulk, first event, Unsubscribe) carries an internal 30 s deadline in addition to the caller token; slots that never publish come back `BadTimeout`, so `CancellationToken.None` can no longer hang on a silent tag.
 - **Supervisory advise is retried after a thrown failure (#605 / 06-S-12)**: the writer marks an item handle advised before the call; a thrown advise (not just a non-OK status) now unmarks it, fails the write closed and re-advises on the next write.

docs/v2/Galaxy.Performance.md

@@ -40,6 +40,9 @@ the metrics surface instead.
 | `galaxy.events.dropped` | `{event}` | MxEvents discarded because the bounded channel was full (newest-dropped) |
 | `galaxy.events.replay_gaps` | `{gap}` | Replay-gap sentinels received — the gateway evicted events we can never see |
 | `galaxy.events.undecodable` | `{event}` | Data-change events carrying a value kind this driver cannot represent |
+| `galaxy.events.early_parked` | `{event}` | Data-change events that overtook their SubscribeBulk reply and were parked for replay (#859) |
+| `galaxy.events.early_replayed` | `{event}` | Parked early events replayed once the reply registered their handle (#859) |
+| `galaxy.events.early_overflow` | `{event}` | Early events dropped because the parking bound (100 000) was reached (#859) |
 
 The invariant is `received = dispatched + dropped + replay_gaps +
 (in-flight in the channel)`. A replay-gap sentinel is counted as

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

@@ -250,9 +250,18 @@ public sealed class GalaxyDriver
     /// <summary>Test seam: true while an <see cref="EventPump"/> is running.</summary>
     internal bool HasEventPumpForTests { get { lock (_pumpLock) return _eventPump is not null; } }
 
+    /// <summary>Test seam — the current pump instance (identity only), to observe a restart.</summary>
+    internal object? EventPumpInstanceForTests { get { lock (_pumpLock) return _eventPump; } }
+
     /// <summary>Test seam: the number of driver subscriptions the registry still tracks.</summary>
     internal int TrackedSubscriptionCountForTests => _subscriptions.TrackedSubscriptionCount;
 
+    /// <summary>Test seam — the registry's parked early-event count (#859).</summary>
+    internal int ParkedEarlyEventCountForTests => _subscriptions.ParkedEventCount;
+
+    /// <summary>Test seam — the driver's subscription registry (#859/#861 window + session tests).</summary>
+    internal SubscriptionRegistry SubscriptionsForTests => _subscriptions;
+
     /// <summary>
     ///     Server-pushed data-change notification. Fires from the
     ///     <see cref="EventPump"/>'s background loop; handlers should be cheap (or queue
@@ -641,19 +650,53 @@ public sealed class GalaxyDriver
     ///     Register <paramref name="bindings"/> only while <paramref name="generation"/> is still the
     ///     live runtime. Atomic with the teardown's registry Clear (both under <see cref="_registryGate"/>).
     /// </summary>
-    private bool TryRegisterForGeneration(long subscriptionId, IReadOnlyList<TagBinding> bindings, int generation)
+    private RegisterOutcome TryRegisterForGeneration(
+        long subscriptionId, IReadOnlyList<TagBinding> bindings, int generation, int session)
     {
         lock (_registryGate)
         {
-            if (Volatile.Read(ref _runtimeGeneration) != generation) return false;
-            _subscriptions.Register(subscriptionId, bindings);
-            return true;
+            if (Volatile.Read(ref _runtimeGeneration) != generation) return RegisterOutcome.TornDown;
+            // #861: the reply may have straddled a reconnect — its handles then belong to the dead
+            // session and collide with the new session's numbers. Checked atomically with
+            // StartNewSession inside the registry.
+            return _subscriptions.TryRegister(subscriptionId, bindings, session)
+                ? RegisterOutcome.Registered
+                : RegisterOutcome.SessionChanged;
+        }
+    }
+
+    private enum RegisterOutcome { Registered, TornDown, SessionChanged }
+
+    /// <summary>
+    ///     UnsubscribeBulk for <paramref name="itemHandles"/>, then purge any event parked for them
+    ///     (#859 I-3 interim — a trailing event must not be promoted onto a reused handle number).
+    /// </summary>
+    private async Task UnsubscribeAndPurgeAsync(
+        IGalaxySubscriber subscriber, IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
+    {
+        // Captured BEFORE the RPC (#859 I-B): a slow or failed unsubscribe can finish after a reconnect,
+        // when these numbers may be the NEW session's — the purge is then a no-op instead of deleting
+        // their parked initial values.
+        var session = _subscriptions.CurrentSession;
+        try
+        {
+            await subscriber.UnsubscribeBulkAsync(itemHandles, cancellationToken).ConfigureAwait(false);
+        }
+        finally
+        {
+            _subscriptions.PurgeParked(itemHandles, session);
         }
     }
 
     private void ClearRegistry()
     {
-        lock (_registryGate) _subscriptions.Clear();
+        lock (_registryGate)
+        {
+            _subscriptions.Clear();
+            // A new session too: a late PurgeParked from a call the teardown orphaned (captured the old
+            // session) must never reach the next runtime's parked values.
+            _subscriptions.StartNewSession();
+        }
     }
 
     /// <summary>
@@ -664,7 +707,7 @@ public sealed class GalaxyDriver
     {
         var live = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
         if (live.Length == 0) return;
-        try { await subscriber.UnsubscribeBulkAsync(live, CancellationToken.None).ConfigureAwait(false); }
+        try { await UnsubscribeAndPurgeAsync(subscriber, live, CancellationToken.None).ConfigureAwait(false); }
         catch (Exception ex)
         {
             _logger.LogWarning(ex,
@@ -918,14 +961,49 @@ public sealed class GalaxyDriver
     /// </summary>
     private async Task ReopenAsync(CancellationToken cancellationToken)
     {
-        if (_ownedMxSession is null) return;
-        var clientOptions = await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false);
-        await _ownedMxSession.RecreateAsync(clientOptions, cancellationToken).ConfigureAwait(false);
+        if (_ownedMxSession is null && RecreateOverrideForTests is null) return;
+
+        // #859 I-C: fence the registry around the swap itself, not later in ReplayAsync — a
+        // SubscribeBulk issued between RecreateAsync and the replay goes to the NEW gateway session.
+        // BEFORE: from here on nothing registered so far counts as live, so an UnsubscribeAsync during
+        // reopen/backoff never sends an old number (which the new session may reuse).
+        _subscriptions.StartNewSession();
+        if (RecreateOverrideForTests is { } recreate)
+        {
+            await recreate(cancellationToken).ConfigureAwait(false);
+        }
+        else
+        {
+            var clientOptions = await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false);
+            await _ownedMxSession!.RecreateAsync(clientOptions, cancellationToken).ConfigureAwait(false);
+        }
+        // AFTER: a subscribe that captured the session between the two bumps may have dialled the old
+        // (dead) session, so it is refused and re-issued; one that captures it from here on dials the
+        // recreated session (GatewayGalaxySubscriber reads GalaxyMxSession.Session at call time), so
+        // it registers normally and the replay leaves it alone instead of re-subscribing it.
+        Volatile.Write(ref _sessionOpenedByReopen, _subscriptions.StartNewSession());
+
         // The recreated session invalidates every prior gw item handle; drop the writer's handle/advise
         // caches so the next write re-AddItems + re-AdviseSupervisory against the fresh session.
         _dataWriter?.InvalidateHandleCaches();
     }
 
+    /// <summary>
+    ///     The registry session <see cref="ReopenAsync"/> opened for the recreated gateway session, or -1.
+    ///     <see cref="ReplayAsync"/> starts a new session itself only when no reopen did (the injected-
+    ///     subscriber path, and tests); a second bump after a reopen would fence subscribes that already
+    ///     registered against the recreated session.
+    /// </summary>
+    private int _sessionOpenedByReopen = -1;
+
+    /// <summary>Test seam — replaces the gateway session swap inside <see cref="ReopenAsync"/>.</summary>
+    internal Func<CancellationToken, Task>? RecreateOverrideForTests { get; set; }
+
+    /// <summary>Test-visible entry into <see cref="ReopenAsync"/>.</summary>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    internal Task InvokeReopenForTestAsync(CancellationToken cancellationToken) => ReopenAsync(cancellationToken);
+
     /// <summary>
     ///     Replay callback. Walks every active subscription, re-issues SubscribeBulk for
     ///     its tag list, and <see cref="SubscriptionRegistry.Rebind">rebinds</see> the
@@ -942,34 +1020,71 @@ public sealed class GalaxyDriver
         // anyway would start a fresh pump underneath the shutdown.
         cancellationToken.ThrowIfCancellationRequested();
         if (_subscriber is null) return;
-        var entries = _subscriptions.SnapshotEntries();
-        if (entries.Count == 0) return;
 
-        // Honor ReplayOnSessionLost. When operators opt out (false) we skip the
-        // per-tag SubscribeBulk fan-out — they're delegating to the gateway's
-        // session-level ReplaySubscriptions or accept post-reconnect tag loss. We
-        // still restart the EventPump so a future Subscribe call lands on a live
-        // consumer.
-        if (!_options.Reconnect.ReplayOnSessionLost)
+        // The session was recreated: every item handle the registry holds is dead and its number
+        // may be reused by the new session (#859 I-1 / #861). Invalidate the reverse map (the
+        // subscriptions stay; the per-iteration Rebind below re-binds each one, and new-session
+        // events park meanwhile instead of resolving to the old owner of a reused number), discard
+        // held events, and fence any subscribe whose reply straddled the reconnect. ReopenAsync
+        // already did this around the session swap; only when no reopen ran (injected subscriber,
+        // tests) is it done here — FIRST, before the pump restart and even with nothing to replay.
+        if (Interlocked.Exchange(ref _sessionOpenedByReopen, -1) != _subscriptions.CurrentSession)
         {
-            RestartEventPumpForReplay();
-            _logger.LogInformation(
-                "GalaxyDriver {InstanceId} reconnect replay skipped — ReplayOnSessionLost=false ({SubCount} subscriptions tracked)",
-                _driverInstanceId, entries.Count);
-            return;
+            _subscriptions.StartNewSession();
         }
 
         // The stream-fault that triggered this recovery left the old pump's RunAsync loop
         // exited and its channel completed; EventPump.Start() is a no-op on a non-null but
-        // completed loop. Recreate the pump so the replayed subscriptions have a consumer.
+        // completed loop. Recreate the pump BEFORE anything can return early (#859 P-1): with no
+        // 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();
 
-        var tagCount = await ResubscribeAllAsync(entries, Volatile.Read(ref _runtimeGeneration), 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).
+        // The probe entry is excluded: ONE registry entry holds every probe handle and any probe publish
+        // on the new session re-stamps it, so its session says nothing about which platforms are stale.
+        // The probe watcher tracks the session per handle and is driven separately below.
+        var session = _subscriptions.CurrentSession;
+        var entries = _subscriptions.SnapshotEntriesNotIn(session)
+            .Where(e => e.SubscriptionId != SubscriptionRegistry.ProbeSubscriptionId)
+            .ToArray();
+        var probeWatcher = _probeWatcher;
+        if (entries.Length == 0 && probeWatcher is null) return;
+
+        // Honor ReplayOnSessionLost. When operators opt out (false) the per-tag SubscribeBulk fan-out
+        // is skipped, and the recreated gateway session carries NONE of the old advises — those
+        // subscriptions stay deaf (their handles are dead and no longer resolve) until re-subscribed.
+        if (!_options.Reconnect.ReplayOnSessionLost)
+        {
+            if (entries.Length > 0)
+            {
+                _logger.LogWarning(
+                    "GalaxyDriver {InstanceId} reconnect replay skipped — ReplayOnSessionLost=false; {SubCount} subscription(s) " +
+                    "receive no further data until they are re-subscribed (the recreated gateway session has none of their advises).",
+                    _driverInstanceId, entries.Length);
+            }
+            return;
+        }
+
+        var generation = Volatile.Read(ref _runtimeGeneration);
+        var tagCount = entries.Length == 0
+            ? 0
+            : await ResubscribeAllAsync(entries, generation, cancellationToken).ConfigureAwait(false);
+
+        // Probes on EVERY reconnect, whatever the registry entry's session: re-subscribes exactly the
+        // platforms whose handles predate the current session (a platform a seed already advised on the
+        // new session is left alone — no double advise), and republishes the fenced bindings.
+        if (probeWatcher is not null
+            && Volatile.Read(ref _runtimeGeneration) == generation
+            && _subscriptions.CurrentSession == session)
+        {
+            tagCount += await probeWatcher.ResubscribeStaleAsync(cancellationToken).ConfigureAwait(false);
+        }
 
         _logger.LogInformation(
             "GalaxyDriver {InstanceId} replay completed — {SubCount} subscriptions, {TagCount} tags re-subscribed",
-            _driverInstanceId, entries.Count, tagCount);
+            _driverInstanceId, entries.Length, tagCount);
     }
 
     /// <summary>
@@ -993,6 +1108,9 @@ public sealed class GalaxyDriver
         CancellationToken cancellationToken)
     {
         var tagCount = 0;
+        // M-b: the session this pass belongs to, read ONCE. Any later change fences the rest of the
+        // pass (the new session's own replay takes over).
+        var session = _subscriptions.CurrentSession;
         foreach (var (subscriptionId, oldBindings) in entries)
         {
             if (Volatile.Read(ref _runtimeGeneration) != generation)
@@ -1000,6 +1118,7 @@ public sealed class GalaxyDriver
                 _logger.LogDebug("GalaxyDriver {InstanceId} re-subscribe abandoned — runtime was torn down.", _driverInstanceId);
                 return tagCount;
             }
+            if (_subscriptions.CurrentSession != session) return tagCount; // a newer session's replay owns the rest
 
             if (subscriptionId == SubscriptionRegistry.ProbeSubscriptionId && _probeWatcher is { } probeWatcher)
             {
@@ -1016,6 +1135,12 @@ public sealed class GalaxyDriver
                 .ToArray();
             if (refs.Length == 0) continue;
 
+            // #859: the re-issued handles' initial values (the whole point of a re-snapshot) can
+            // overtake this SubscribeBulk reply; park them until the Rebind below promotes them.
+            // Scoped to this one iteration (the using is disposed at the end of the loop body): a
+            // window held across all N round-trips would park every unrelated unknown-handle event
+            // for the whole replay and could exhaust the bound before the genuine early values.
+            using var earlyEventWindow = _subscriptions.BeginPendingSubscribe();
             var results = await _subscriber!
                 .SubscribeBulkAsync(refs, _options.MxAccess.PublishingIntervalMs, cancellationToken)
                 .ConfigureAwait(false);
@@ -1033,10 +1158,41 @@ public sealed class GalaxyDriver
             // Rebind so the EventPump fan-out reverse map points at the post-reconnect
             // handles; otherwise events on the new handles miss every subscription. Not after a
             // teardown: the fresh handles belong to a subscriber that is already being released.
+            SubscriptionRegistry.BindOutcome outcome;
             lock (_registryGate)
             {
                 if (Volatile.Read(ref _runtimeGeneration) != generation) return tagCount;
-                _subscriptions.Rebind(subscriptionId, newBindings);
+                outcome = _subscriptions.TryRebind(subscriptionId, newBindings, session);
+            }
+
+            if (outcome == SubscriptionRegistry.BindOutcome.SessionChanged)
+            {
+                // #861: another reconnect during the pass makes these handles dead too; the newer
+                // session's own replay re-subscribes everything, so stop here.
+                _logger.LogDebug(
+                    "GalaxyDriver {InstanceId} re-subscribe abandoned — the gateway session changed again.",
+                    _driverInstanceId);
+                return tagCount;
+            }
+            if (outcome == SubscriptionRegistry.BindOutcome.UnknownId)
+            {
+                // M-a: the subscription (or a subscribe-once read) was removed while its SubscribeBulk
+                // was in flight. Nothing will ever unsubscribe what was just advised — release it now.
+                var orphaned = newBindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
+                if (orphaned.Length > 0)
+                {
+                    try
+                    {
+                        await UnsubscribeAndPurgeAsync(_subscriber!, orphaned, CancellationToken.None).ConfigureAwait(false);
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogWarning(ex,
+                            "GalaxyDriver {InstanceId} could not release {Count} handle(s) re-subscribed for a subscription removed during the pass.",
+                            _driverInstanceId, orphaned.Length);
+                    }
+                }
+                continue;
             }
             tagCount += refs.Length;
         }
@@ -1392,6 +1548,13 @@ public sealed class GalaxyDriver
         return ReadViaSubscribeOnceAsync(dialledRefs, cancellationToken);
     }
 
+    /// <summary>
+    ///     Raw (attributeRef-domain) data changes for in-flight subscribe-once reads. Raised by
+    ///     <see cref="OnPumpDataChange"/> for whichever pump is current, so a read survives a reconnect's
+    ///     pump restart.
+    /// </summary>
+    private event EventHandler<DataChangeEventArgs>? ReadDataChange;
+
     /// <summary>Upper bound on one subscribe-once read; internal so tests can shrink it.</summary>
     internal TimeSpan ReadViaSubscribeDeadline { get; set; } = TimeSpan.FromSeconds(30);
 
@@ -1420,7 +1583,9 @@ public sealed class GalaxyDriver
 
         var generation = Volatile.Read(ref _runtimeGeneration); // see SubscribeAsync
         var readSubscriber = _subscriber!;
-        var pump = EnsureEventPumpStarted();
+        // 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 subscriptionId = _subscriptions.NextSubscriptionId();
 
         // Pre-allocate one TaskCompletionSource per full-reference so the OnDataChange
@@ -1446,79 +1611,114 @@ public sealed class GalaxyDriver
                 tcs.TrySetResult(args.Snapshot);
             }
         };
-        pump.OnDataChange += handler;
+        ReadDataChange += handler;
 
         var bufferedIntervalMs = _options.MxAccess.PublishingIntervalMs;
         IReadOnlyList<SubscribeResult> results;
-        // The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow
-        // gateway can never surface an OperationCanceledException to a caller that did not cancel, and the
-        // handles it eventually creates are known (and unsubscribed below) rather than lost to a cancelled call.
-        var subscribeTask = readSubscriber.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken);
+        List<TagBinding> bindings;
+
+        // #859: a read waits ONLY for the initial value, and that value can overtake the SubscribeBulk
+        // reply — dropped as stale, a slow-changing attribute would read BadTimeout. Open the early-event
+        // window before the RPC and close it right after registration (the finally below), not after the
+        // value wait: a window held for the whole wait would park every unrelated unknown-handle event.
+        var earlyEventWindow = _subscriptions.BeginPendingSubscribe();
+        var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on
         try
         {
-            results = await subscribeTask.WaitAsync(cancellationToken).ConfigureAwait(false);
-        }
-        catch (OperationCanceledException) when (!callerToken.IsCancellationRequested)
-        {
-            // Internal deadline fired while the subscribe RPC was still in flight.
-            pump.OnDataChange -= handler;
-            var subscriber = readSubscriber;
-            var logger = _logger;
-            _ = subscribeTask.ContinueWith(t =>
+            // The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow
+            // gateway can never surface an OperationCanceledException to a caller that did not cancel, and the
+            // handles it eventually creates are known (and unsubscribed below) rather than lost to a cancelled call.
+            var subscribeTask = readSubscriber.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken);
+            try
             {
-                if (!t.IsCompletedSuccessfully) return Task.CompletedTask;
-                var late = t.Result.Where(r => r.WasSuccessful && r.ItemHandle > 0).Select(r => r.ItemHandle).ToArray();
-                return late.Length == 0
-                    ? Task.CompletedTask
-                    : subscriber.UnsubscribeBulkAsync(late, CancellationToken.None).ContinueWith(
-                        u => logger.LogWarning(u.Exception,
-                            "GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length),
-                        TaskContinuationOptions.OnlyOnFaulted);
-            }, TaskScheduler.Default).Unwrap();
-            var now = DateTime.UtcNow;
-            var timedOut = new DataValueSnapshot[fullReferences.Count];
-            for (var i = 0; i < timedOut.Length; i++)
-                timedOut[i] = new DataValueSnapshot(null, StatusCodeMap.BadTimeout, null, now);
-            return timedOut;
-        }
-        catch
-        {
-            pump.OnDataChange -= handler;
-            throw;
-        }
-
-        // Register bindings so the pump knows to dispatch events for these handles.
-        // Index the SubscribeBulk results once and correlate to references in O(1)
-        // instead of FirstOrDefault per element (O(n²) over the batch). On the
-        // 50k-tag soak path this turns a 2.5G-comparison loop into a single
-        // Dictionary build + linear scan.
-        var resultIndex = BuildResultIndex(results);
-        var bindings = new List<TagBinding>(fullReferences.Count);
-        for (var i = 0; i < fullReferences.Count; i++)
-        {
-            var fullRef = fullReferences[i];
-            var itemHandle = resultIndex.TryGetValue(fullRef, out var match) && match is { WasSuccessful: true }
-                ? match.ItemHandle
-                : 0;
-            bindings.Add(new TagBinding(fullRef, itemHandle));
-
-            // Tags the gw rejected up front — complete with Bad status now so the
-            // wait below doesn't time out on them.
-            if (itemHandle <= 0
-                && pendingByRef.TryGetValue(fullRef, out var rejectedTcs))
+                results = await subscribeTask.WaitAsync(cancellationToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (!callerToken.IsCancellationRequested)
             {
-                rejectedTcs.TrySetResult(new DataValueSnapshot(
-                    Value: null,
-                    StatusCode: StatusCodeMap.Bad,
-                    SourceTimestampUtc: null,
-                    ServerTimestampUtc: DateTime.UtcNow));
+                // Internal deadline fired while the subscribe RPC was still in flight.
+                ReadDataChange -= handler;
+                var subscriber = readSubscriber;
+                var logger = _logger;
+                _ = subscribeTask.ContinueWith(t =>
+                {
+                    if (!t.IsCompletedSuccessfully) return Task.CompletedTask;
+                    // A late reply from a session that has since been replaced carries dead handle
+                    // numbers the new session may reuse — never unsubscribe those (#861).
+                    if (_subscriptions.CurrentSession != session) return Task.CompletedTask;
+                    var late = t.Result.Where(r => r.WasSuccessful && r.ItemHandle > 0).Select(r => r.ItemHandle).ToArray();
+                    return late.Length == 0
+                        ? Task.CompletedTask
+                        : UnsubscribeAndPurgeAsync(subscriber, late, CancellationToken.None).ContinueWith(
+                            u => logger.LogWarning(u.Exception,
+                                "GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length),
+                            TaskContinuationOptions.OnlyOnFaulted);
+                }, TaskScheduler.Default).Unwrap();
+                var now = DateTime.UtcNow;
+                var timedOut = new DataValueSnapshot[fullReferences.Count];
+                for (var i = 0; i < timedOut.Length; i++)
+                    timedOut[i] = new DataValueSnapshot(null, StatusCodeMap.BadTimeout, null, now);
+                return timedOut;
+            }
+            catch
+            {
+                ReadDataChange -= handler;
+                throw;
+            }
+
+            // Register bindings so the pump knows to dispatch events for these handles.
+            // Index the SubscribeBulk results once and correlate to references in O(1)
+            // instead of FirstOrDefault per element (O(n²) over the batch). On the
+            // 50k-tag soak path this turns a 2.5G-comparison loop into a single
+            // Dictionary build + linear scan.
+            var resultIndex = BuildResultIndex(results);
+            bindings = new List<TagBinding>(fullReferences.Count);
+            for (var i = 0; i < fullReferences.Count; i++)
+            {
+                var fullRef = fullReferences[i];
+                var itemHandle = resultIndex.TryGetValue(fullRef, out var match) && match is { WasSuccessful: true }
+                    ? match.ItemHandle
+                    : 0;
+                bindings.Add(new TagBinding(fullRef, itemHandle));
+
+                // Tags the gw rejected up front — complete with Bad status now so the
+                // wait below doesn't time out on them.
+                if (itemHandle <= 0
+                    && pendingByRef.TryGetValue(fullRef, out var rejectedTcs))
+                {
+                    rejectedTcs.TrySetResult(new DataValueSnapshot(
+                        Value: null,
+                        StatusCode: StatusCodeMap.Bad,
+                        SourceTimestampUtc: null,
+                        ServerTimestampUtc: DateTime.UtcNow));
+                }
+            }
+            var registered = TryRegisterForGeneration(subscriptionId, bindings, generation, session);
+            if (registered != RegisterOutcome.Registered)
+            {
+                ReadDataChange -= handler;
+                earlyEventWindow.Dispose(); // before any further round-trip
+                if (registered == RegisterOutcome.TornDown)
+                {
+                    await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false);
+                    throw TornDownDuringSubscribe();
+                }
+
+                // #861: the reply straddled a reconnect. Its handles died with the old session and
+                // their numbers may already belong to new-session items, so they are neither
+                // registered nor unsubscribed. A read is one-shot: report the lost session per tag.
+                _logger.LogInformation(
+                    "GalaxyDriver {InstanceId} subscribe-once read of {Count} tag(s) straddled a gateway reconnect — returning BadCommunicationError.",
+                    _driverInstanceId, fullReferences.Count);
+                var lostAt = DateTime.UtcNow;
+                var lost = new DataValueSnapshot[fullReferences.Count];
+                for (var i = 0; i < lost.Length; i++)
+                    lost[i] = new DataValueSnapshot(null, StatusCodeMap.BadCommunicationError, null, lostAt);
+                return lost;
             }
         }
-        if (!TryRegisterForGeneration(subscriptionId, bindings, generation))
+        finally
         {
-            pump.OnDataChange -= handler;
-            await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false);
-            throw TornDownDuringSubscribe();
+            earlyEventWindow.Dispose();
         }
 
         try
@@ -1548,17 +1748,24 @@ public sealed class GalaxyDriver
         }
         finally
         {
-            pump.OnDataChange -= handler;
+            ReadDataChange -= handler;
             // Drop the bindings + unsubscribe the live handles. UnsubscribeBulkAsync's
             // failure isn't fatal — the registry is already cleared, so any straggling
             // event from the gw would be a no-op fan-out.
-            _subscriptions.Remove(subscriptionId);
-            var liveHandles = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
+            var removed = _subscriptions.Remove(subscriptionId, out var liveInCurrentSession);
+            // Unsubscribe what the registry held at removal, NOT the local `bindings` (#859 B-1): a
+            // reconnect replay during the wait may have rebound this read to new-session handles, and
+            // the original numbers may now be other tags' live items. If the entry still holds dead
+            // old-session handles (not yet rebound), release nothing (#861). Only when the entry was
+            // already gone (teardown cleared the registry) fall back to the original bindings.
+            IReadOnlyList<TagBinding> source = removed ?? bindings;
+            if (removed is not null && !liveInCurrentSession) source = [];
+            var liveHandles = source.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
             if (liveHandles.Length > 0)
             {
                 try
                 {
-                    await readSubscriber.UnsubscribeBulkAsync(liveHandles, CancellationToken.None)
+                    await UnsubscribeAndPurgeAsync(readSubscriber, liveHandles, CancellationToken.None)
                         .ConfigureAwait(false);
                 }
                 catch (Exception ex)
@@ -1630,7 +1837,14 @@ public sealed class GalaxyDriver
         {
             // Empty subscriptions register but never bind anything — keeps Unsubscribe
             // symmetric for callers that conditionally add tags later.
-            if (!TryRegisterForGeneration(subscriptionId, [], generation)) throw TornDownDuringSubscribe();
+            RegisterOutcome outcome;
+            do
+            {
+                // No handles, so a session change only means "read the session again".
+                outcome = TryRegisterForGeneration(subscriptionId, [], generation, _subscriptions.CurrentSession);
+            }
+            while (outcome == RegisterOutcome.SessionChanged);
+            if (outcome == RegisterOutcome.TornDown) throw TornDownDuringSubscribe();
             return new GalaxySubscriptionHandle(subscriptionId);
         }
 
@@ -1646,10 +1860,64 @@ public sealed class GalaxyDriver
         // and infrastructure callers (probe watcher, deploy watcher) hit the fallback.
         var requested = (int)Math.Max(0, publishingInterval.TotalMilliseconds);
         var bufferedIntervalMs = requested > 0 ? requested : _options.MxAccess.PublishingIntervalMs;
-        var results = await subscriber
-            .SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken)
-            .ConfigureAwait(false);
 
+        for (var attempt = 1; ; attempt++)
+        {
+            // #859: MXAccess's initial value for a new handle can reach the event pump before this
+            // reply is registered (the gateway gives no reply-before-event ordering, and the reply and
+            // the event ride separate RPCs). Open the early-event window BEFORE the RPC so such events
+            // are parked rather than dropped as stale; it closes once the registration (which promotes
+            // them for in-order replay) is done.
+            var earlyEventWindow = _subscriptions.BeginPendingSubscribe();
+            try
+            {
+                var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on
+                var results = await subscriber
+                    .SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken)
+                    .ConfigureAwait(false);
+                var bindings = BuildSubscribeBindings(dialledRefs, results);
+
+                switch (TryRegisterForGeneration(subscriptionId, bindings, generation, session))
+                {
+                    case RegisterOutcome.Registered:
+                        _ = pump; // keep the pump alive for the subscription's lifetime
+                        return new GalaxySubscriptionHandle(subscriptionId);
+
+                    case RegisterOutcome.TornDown:
+                        // The teardown cleared the registry while SubscribeBulk was in flight; registering
+                        // now would carry dead handles into the next runtime. Release what we got and fail.
+                        earlyEventWindow.Dispose(); // M-3: not held across the release round-trip
+                        await ReleaseOrphanedHandlesAsync(subscriber, bindings).ConfigureAwait(false);
+                        throw TornDownDuringSubscribe();
+
+                    default:
+                        // #861: the reply straddled a gateway reconnect, so its handles died with the old
+                        // session (and are NOT unsubscribed — the new session may reuse the numbers). The
+                        // reconnect replay only re-subscribes what was registered when it snapshotted, so
+                        // this subscription would be lost: issue it again on the new session.
+                        if (attempt >= MaxSubscribeSessionAttempts)
+                        {
+                            throw new InvalidOperationException(
+                                $"Galaxy subscribe straddled {attempt} consecutive gateway reconnects — giving up; retry later.");
+                        }
+                        _logger.LogInformation(
+                            "GalaxyDriver {InstanceId} subscribe of {Count} tag(s) straddled a gateway reconnect — re-issuing on the new session (attempt {Attempt}).",
+                            _driverInstanceId, dialledRefs.Length, attempt + 1);
+                        continue;
+                }
+            }
+            finally
+            {
+                earlyEventWindow.Dispose();
+            }
+        }
+    }
+
+    /// <summary>Bound on re-issuing a subscribe whose reply straddled a gateway reconnect (#861).</summary>
+    private const int MaxSubscribeSessionAttempts = 3;
+
+    private List<TagBinding> BuildSubscribeBindings(string[] dialledRefs, IReadOnlyList<SubscribeResult> results)
+    {
         // Build the binding list in input order. Failed entries (gw rejected the tag) are
         // recorded with a non-positive ItemHandle so the caller can detect partial failure
         // by inspecting the returned handle's diagnostic context — full per-tag error
@@ -1673,16 +1941,7 @@ public sealed class GalaxyDriver
                     fullRef, match?.ErrorMessage ?? "<no result returned>");
             }
         }
-
-        if (!TryRegisterForGeneration(subscriptionId, bindings, generation))
-        {
-            // The teardown cleared the registry while SubscribeBulk was in flight; registering now
-            // would carry dead handles into the next runtime. Release what we got and fail the call.
-            await ReleaseOrphanedHandlesAsync(subscriber, bindings).ConfigureAwait(false);
-            throw TornDownDuringSubscribe();
-        }
-        _ = pump; // keep the pump alive for the subscription's lifetime
-        return new GalaxySubscriptionHandle(subscriptionId);
+        return bindings;
     }
 
     /// <inheritdoc />
@@ -1697,15 +1956,19 @@ public sealed class GalaxyDriver
                 nameof(handle));
         }
 
-        var bindings = _subscriptions.Remove(gsh.SubscriptionId);
+        var bindings = _subscriptions.Remove(gsh.SubscriptionId, out var liveInCurrentSession);
         if (bindings is null) return; // already removed or never registered
 
+        // Not yet rebound after a reconnect: its handles died with the old session and the new one may
+        // have reused the numbers for other items — unsubscribing them would unadvise those (#861).
+        if (!liveInCurrentSession) return;
+
         var liveItemHandles = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
         if (liveItemHandles.Length == 0 || _subscriber is null) return;
 
         try
         {
-            await _subscriber.UnsubscribeBulkAsync(liveItemHandles, cancellationToken).ConfigureAwait(false);
+            await UnsubscribeAndPurgeAsync(_subscriber, liveItemHandles, cancellationToken).ConfigureAwait(false);
         }
         catch (Exception ex)
         {
@@ -2129,6 +2392,14 @@ public sealed class GalaxyDriver
         // supervisor Healthy transition, so a 90 s Degraded tripped a re-init under flowing data).
         StampLiveness();
 
+        // Subscribe-once reads listen here (attributeRef domain), not on a pump instance: a reconnect
+        // replaces the pump, and a read attached to the dead one never saw the replayed value (#859 B-1).
+        try { ReadDataChange?.Invoke(this, args); }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "GalaxyDriver {InstanceId} read listener threw — continuing.", _driverInstanceId);
+        }
+
         // Probe traffic rides the reserved registry entry and stays in the attributeRef domain —
         // ScanState attributes are discovered platform attributes, never authored RawPaths. It is
         // the watcher's alone: the host never subscribed it and must not see it on OnDataChange.

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs

@@ -46,7 +46,10 @@ public sealed class PerPlatformProbeWatcher : IDisposable
 
     // Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe;
     // we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call.
-    private readonly ConcurrentDictionary<string, int> _itemHandlesByPlatform =
+    // Each handle carries the registry session that issued it (#859 I-A): item handles are per gateway
+    // session and the next session reuses the numbers, so an old-session handle must never be published,
+    // unsubscribed or purged against the current session.
+    private readonly ConcurrentDictionary<string, ProbeHandle> _itemHandlesByPlatform =
         new(StringComparer.OrdinalIgnoreCase);
     private readonly Lock _syncLock = new();
     // Written under _syncLock by StopAsync; read under it wherever a write-back must not outlive a stop.
@@ -99,7 +102,14 @@ public sealed class PerPlatformProbeWatcher : IDisposable
     internal IReadOnlyList<TagBinding> CurrentBindings => BuildBindings();
 
     private IReadOnlyList<TagBinding> BuildBindings() =>
-        [.. _itemHandlesByPlatform.Select(kvp => new TagBinding(kvp.Key + ProbeSuffix, kvp.Value))];
+        [.. _itemHandlesByPlatform.Select(kvp => new TagBinding(kvp.Key + ProbeSuffix, kvp.Value.ItemHandle))];
+
+    /// <summary>The bindings to publish for <paramref name="session"/>: another session's handles bind as 0 (no events).</summary>
+    private IReadOnlyList<TagBinding> BuildBindingsFor(int session) =>
+        [.. _itemHandlesByPlatform.Select(kvp => new TagBinding(
+            kvp.Key + ProbeSuffix, kvp.Value.Session == session ? kvp.Value.ItemHandle : 0))];
+
+    private int CurrentSession => _registry?.CurrentSession ?? 0;
 
     /// <summary>
     ///     Mirror the tracked handle set into the driver registry (no-op without one, or once stopped —
@@ -107,15 +117,20 @@ public sealed class PerPlatformProbeWatcher : IDisposable
     /// </summary>
     private void PublishBindings()
     {
-        lock (_syncLock) PublishBindingsLocked();
+        lock (_syncLock) PublishBindingsLocked(CurrentSession);
     }
 
-    private void PublishBindingsLocked()
+    /// <returns>False when the registry's session is no longer <paramref name="session"/> (nothing published).</returns>
+    private bool PublishBindingsLocked(int session)
     {
-        if (_registry is null || _disposed) return;
-        var bindings = BuildBindings();
-        if (bindings.Count == 0) _registry.Remove(SubscriptionRegistry.ProbeSubscriptionId);
-        else _registry.Upsert(SubscriptionRegistry.ProbeSubscriptionId, bindings);
+        if (_registry is null || _disposed) return true;
+        if (_itemHandlesByPlatform.IsEmpty)
+        {
+            _registry.Remove(SubscriptionRegistry.ProbeSubscriptionId);
+            return true;
+        }
+        // Fenced (#859 I-A): only handles issued by `session` are bound, and only while it is current.
+        return _registry.TryUpsert(SubscriptionRegistry.ProbeSubscriptionId, BuildBindingsFor(session), session);
     }
 
     /// <summary>
@@ -135,9 +150,56 @@ public sealed class PerPlatformProbeWatcher : IDisposable
         return platforms.Length;
     }
 
+    /// <summary>
+    ///     The reconnect replay: re-issue SubscribeBulk only for platforms whose handle predates the
+    ///     registry's current session, then republish. A platform a seed already advised on the new
+    ///     session is left alone (re-subscribing it would double-advise and leak a handle), and every
+    ///     reconnect calls this regardless of the shared registry entry's session — that entry is
+    ///     re-stamped by any probe publish, so it cannot say which platforms are stale (#859).
+    /// </summary>
+    /// <param name="cancellationToken">A cancellation token to cancel the operation.</param>
+    /// <returns>The number of probe references re-subscribed.</returns>
+    public async Task<int> ResubscribeStaleAsync(CancellationToken cancellationToken)
+    {
+        ObjectDisposedException.ThrowIf(_disposed, this);
+        string[] stale;
+        lock (_syncLock)
+        {
+            var session = CurrentSession;
+            stale = [.. _itemHandlesByPlatform.Where(kvp => kvp.Value.Session != session).Select(kvp => kvp.Key)];
+        }
+        if (stale.Length == 0) return 0;
+
+        await SubscribeAndRecordAsync(stale, cancellationToken).ConfigureAwait(false);
+        return stale.Length;
+    }
+
     private async Task SubscribeAndRecordAsync(IReadOnlyList<string> platforms, CancellationToken cancellationToken)
+    {
+        // #861 / #859 I-A: a reply that straddled a gateway reconnect carries dead handles; re-issue.
+        for (var attempt = 1; ; attempt++)
+        {
+            if (await TrySubscribeAndRecordAsync(platforms, cancellationToken).ConfigureAwait(false)) return;
+            if (attempt >= MaxSessionAttempts)
+            {
+                _logger.LogWarning(
+                    "PerPlatformProbeWatcher probe subscribe straddled {Attempts} gateway reconnects — giving up until the next replay/seed", attempt);
+                return;
+            }
+        }
+    }
+
+    private const int MaxSessionAttempts = 3;
+
+    /// <returns>False when the reply straddled a session change (nothing recorded; retry).</returns>
+    private async Task<bool> TrySubscribeAndRecordAsync(IReadOnlyList<string> platforms, CancellationToken cancellationToken)
     {
         var probeAddresses = platforms.Select(p => p + ProbeSuffix).ToArray();
+
+        // #859: the initial ScanState value can overtake the SubscribeBulk reply; park it until
+        // PublishBindingsLocked registers the handles below (window closes at method exit).
+        using var earlyEventWindow = _registry?.BeginPendingSubscribe();
+        var session = CurrentSession; // the session this SubscribeBulk is issued on
         // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence
         // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState
         // changes are rare so a coarser interval is usually fine; deployments that need
@@ -163,8 +225,15 @@ public sealed class PerPlatformProbeWatcher : IDisposable
             stopped = _disposed;
             if (!stopped)
             {
-                foreach (var (platform, itemHandle, _) in acquired) _itemHandlesByPlatform[platform] = itemHandle;
-                PublishBindingsLocked();
+                if (_registry is not null && _registry.CurrentSession != session)
+                {
+                    // Straddled a reconnect: these handles died with the old session and their numbers may
+                    // be reused — neither record, publish nor unsubscribe them.
+                    return false;
+                }
+                foreach (var (platform, itemHandle, _) in acquired)
+                    _itemHandlesByPlatform[platform] = new ProbeHandle(itemHandle, session);
+                if (!PublishBindingsLocked(session)) return false;
             }
         }
 
@@ -180,7 +249,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable
                         "PerPlatformProbeWatcher could not release {Count} probe handle(s) acquired after stop", orphans.Length);
                 }
             }
-            return;
+            return true;
         }
 
         foreach (var (platform, itemHandle, error) in acquired)
@@ -190,6 +259,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable
                 _logger.LogWarning("PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}", platform, error);
             }
         }
+        return true;
     }
 
     /// <summary>
@@ -224,12 +294,29 @@ public sealed class PerPlatformProbeWatcher : IDisposable
         // race on the membership view.
         List<string> toAdd;
         List<(string Platform, int ItemHandle)> toRemove;
+        int removeSession;
+        bool droppedWithoutRelease;
         lock (_syncLock)
         {
             toAdd = [.. desired.Where(p => !_itemHandlesByPlatform.ContainsKey(p))];
+            // Captured before the RPC (#859 I-B); only handles issued by the CURRENT session are released —
+            // an older session's numbers may now belong to unrelated live items (#859 I-A).
+            removeSession = CurrentSession;
             toRemove = [.. _itemHandlesByPlatform
-                .Where(kvp => !desired.Contains(kvp.Key) && kvp.Value > 0)
-                .Select(kvp => (kvp.Key, kvp.Value))];
+                .Where(kvp => !desired.Contains(kvp.Key)
+                              && kvp.Value.ItemHandle > 0 && kvp.Value.Session == removeSession)
+                .Select(kvp => (kvp.Key, kvp.Value.ItemHandle))];
+            var deadRemovals = _itemHandlesByPlatform
+                .Where(kvp => !desired.Contains(kvp.Key) && !(kvp.Value.ItemHandle > 0 && kvp.Value.Session == removeSession))
+                .Select(kvp => kvp.Key).ToArray();
+            // Dropped platforms with no releasable handle (rejected, or an older session's dead number):
+            // forget them without any unsubscribe.
+            foreach (var platform in deadRemovals)
+            {
+                _itemHandlesByPlatform.TryRemove(platform, out _);
+                _aggregator.Remove(platform);
+            }
+            droppedWithoutRelease = deadRemovals.Length > 0;
 
             // Drop removed entries from the membership map up-front so a concurrent
             // OnProbeValueChanged for them is silently ignored. The unsubscribe RPC
@@ -241,6 +328,8 @@ public sealed class PerPlatformProbeWatcher : IDisposable
             }
         }
 
+        if (droppedWithoutRelease && toRemove.Count == 0) PublishBindings(); // keep the fan-out map in step
+
         if (toRemove.Count > 0)
         {
             // Drop the dead handles from the fan-out map before the (slow, fallible) unsubscribe.
@@ -256,6 +345,12 @@ public sealed class PerPlatformProbeWatcher : IDisposable
                     "PerPlatformProbeWatcher unsubscribe failed for {Count} probe(s); aggregator entries already cleared.",
                     toRemove.Count);
             }
+            finally
+            {
+                // #859 I-3 interim: no trailing event for a removed probe handle may be promoted onto
+                // a reused handle number.
+                _registry?.PurgeParked([.. toRemove.Select(t => t.ItemHandle)], removeSession);
+            }
         }
 
         if (toAdd.Count == 0) return;
@@ -324,7 +419,9 @@ public sealed class PerPlatformProbeWatcher : IDisposable
         {
             if (_disposed) return;
             _disposed = true;
-            liveHandles = [.. _itemHandlesByPlatform.Values.Where(h => h > 0)];
+            var session = CurrentSession; // never release an older session's (reusable) numbers
+            liveHandles = [.. _itemHandlesByPlatform.Values
+                .Where(h => h.ItemHandle > 0 && h.Session == session).Select(h => h.ItemHandle)];
             _itemHandlesByPlatform.Clear();
             _registry?.Remove(SubscriptionRegistry.ProbeSubscriptionId);
         }
@@ -337,4 +434,9 @@ public sealed class PerPlatformProbeWatcher : IDisposable
 
     /// <summary>Disposes the probe watcher and unsubscribes all tracked platforms (blocking on the round-trip).</summary>
     public void Dispose() => StopAsync(CancellationToken.None).GetAwaiter().GetResult();
+
+    /// <summary>A probe item handle and the registry session that issued it.</summary>
+    /// <param name="ItemHandle">The gateway item handle (0 = rejected).</param>
+    /// <param name="Session">The registry session the handle belongs to.</param>
+    private readonly record struct ProbeHandle(int ItemHandle, int Session);
 }

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

@@ -53,6 +53,15 @@ internal sealed class EventPump : IAsyncDisposable
     private static readonly Counter<long> ValuesUndecodable =
         Meter.CreateCounter<long>("galaxy.events.undecodable", unit: "{event}",
             description: "Data-change events carrying a value kind this driver cannot represent.");
+    private static readonly Counter<long> EarlyEventsParked =
+        Meter.CreateCounter<long>("galaxy.events.early_parked", unit: "{event}",
+            description: "Data-change events that arrived before their SubscribeBulk reply registered the item handle, parked for replay (#859).");
+    private static readonly Counter<long> EarlyEventsReplayed =
+        Meter.CreateCounter<long>("galaxy.events.early_replayed", unit: "{event}",
+            description: "Parked early data-change events replayed once their item handle was registered (#859).");
+    private static readonly Counter<long> EarlyEventsOverflowed =
+        Meter.CreateCounter<long>("galaxy.events.early_overflow", unit: "{event}",
+            description: "Early data-change events dropped because the parking bound was reached (#859).");
 
     private readonly IGalaxySubscriber _subscriber;
     private readonly SubscriptionRegistry _registry;
@@ -124,11 +133,38 @@ internal sealed class EventPump : IAsyncDisposable
             // value consistent).
             FullMode = BoundedChannelFullMode.Wait,
             SingleReader = true,
-            SingleWriter = true,
+            // Not single-writer: besides the stream read loop, a registration that promotes
+            // parked early events (#859) writes the WakeSentinel from the subscribing thread.
+            SingleWriter = false,
         });
         _clientTag = new KeyValuePair<string, object?>("galaxy.client", clientName ?? "<unknown>");
+
+        // One pump per gateway session. The owner calls SubscriptionRegistry.StartNewSession() BEFORE
+        // creating the pump for a new session (reconnect); this pump binds to that session and never
+        // parks or replays under any other (#859 review — item handles are reused across sessions).
+        _session = _registry.CurrentSession;
+        _registry.ParkedEventsReady += WakeForReadyEvents;
     }
 
+    private readonly int _session;
+
+    /// <summary>
+    ///     Marker written into the channel to wake the dispatch loop after a registration promoted
+    ///     parked early events (#859). Never dispatched itself; compared by reference.
+    /// </summary>
+    private static readonly MxEvent WakeSentinel = new();
+
+    /// <summary>
+    ///     Wakes the dispatch loop so promoted early events replay even if no further event arrives.
+    ///     Runs on the registering thread, possibly under the driver's <c>_registryGate</c> or the probe
+    ///     watcher's <c>_syncLock</c> — safe because the channel keeps the default
+    ///     <c>AllowSynchronousContinuations = false</c>, so the write never runs the dispatch loop inline.
+    ///     A failed write is harmless: the channel is either full (the loop is busy and replays the
+    ///     ready queue before its next dispatch) or completed (the pump is dead; its held events
+    ///     belong to a session the next StartNewSession discards).
+    /// </summary>
+    private void WakeForReadyEvents() => _channel.Writer.TryWrite(WakeSentinel);
+
     /// <summary>
     ///     Start consuming the event stream on a background task. Idempotent — second
     ///     calls are no-ops while the loop is running.
@@ -281,6 +317,11 @@ internal sealed class EventPump : IAsyncDisposable
         {
             await foreach (var ev in _channel.Reader.ReadAllAsync(ct).ConfigureAwait(false))
             {
+                if (ReferenceEquals(ev, WakeSentinel))
+                {
+                    DispatchReadyEvents();
+                    continue;
+                }
                 Dispatch(ev);
                 EventsDispatched.Add(1, _clientTag);
             }
@@ -313,16 +354,64 @@ internal sealed class EventPump : IAsyncDisposable
                 // — alarms come from the gateway's session-less StreamAlarms feed
                 // (GatewayGalaxyAlarmFeed). OnWriteComplete / OperationComplete /
                 // OnBufferedDataChange are filtered out: write callers get their reply
-                // via the InvokeAsync round-trip, not via the event stream.
+                // via the InvokeAsync round-trip, not via the event stream. Still replay any
+                // promoted early events: a wake lost to a full channel relies on the next item.
+                DispatchReadyEvents();
                 return;
         }
     }
 
     private void DispatchDataChange(MxEvent ev)
     {
-        var subscribers = _registry.ResolveSubscribers(ev.ItemHandle);
-        if (subscribers.Count == 0) return; // stale event after unsubscribe — drop quietly
+        var outcome = _registry.ResolveOrPark(ev, _session, out var subscribers, out var replayFirst);
 
+        // Replay promoted early events (#859) BEFORE fanning this event out. A registration writes
+        // the reverse map and promotes its parked events in ONE critical section under the
+        // registry's park gate, and ResolveOrPark resolves — and takes the ready queue — under that
+        // same gate whenever anything is held. So if this event resolved, every older parked event
+        // for its handle is in replayFirst, and replaying it first preserves per-handle order.
+        ReplayEarly(replayFirst);
+
+        switch (outcome)
+        {
+            case EarlyEventOutcome.Resolved:
+                break;
+            case EarlyEventOutcome.Parked:
+                // The handle's SubscribeBulk reply has not been registered yet — MXAccess's initial
+                // value can overtake it. Replayed when the registration lands.
+                EarlyEventsParked.Add(1, _clientTag);
+                return;
+            case EarlyEventOutcome.Overflow:
+                EarlyEventsOverflowed.Add(1, _clientTag);
+                return;
+            default:
+                return; // stale event after unsubscribe (no subscribe in flight) — drop quietly
+        }
+
+        FanOut(ev, subscribers);
+    }
+
+    /// <summary>
+    ///     Dispatch every early event a registration promoted (#859). Runs only on the dispatch
+    ///     thread, so it is serialised with live dispatch. A promoted event whose subscription was
+    ///     removed again before the replay resolves to nobody and is dropped like any stale event.
+    /// </summary>
+    private void DispatchReadyEvents() => ReplayEarly(_registry.TakeReadyEvents(_session));
+
+    private void ReplayEarly(IReadOnlyList<MxEvent> ready)
+    {
+        if (ready.Count == 0) return;
+        foreach (var early in ready)
+        {
+            var subscribers = _registry.ResolveSubscribers(early.ItemHandle);
+            if (subscribers.Count == 0) continue;
+            EarlyEventsReplayed.Add(1, _clientTag);
+            FanOut(early, subscribers);
+        }
+    }
+
+    private void FanOut(MxEvent ev, IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers)
+    {
         var snapshot = ToSnapshot(ev);
         foreach (var (subscriptionId, fullReference) in subscribers)
         {
@@ -377,6 +466,7 @@ internal sealed class EventPump : IAsyncDisposable
     {
         if (_disposed) return;
         _disposed = true;
+        _registry.ParkedEventsReady -= WakeForReadyEvents;
         _cts.Cancel();
         _channel.Writer.TryComplete();
         if (_loop is not null)

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

@@ -1,5 +1,6 @@
 using System.Collections.Concurrent;
 using System.Collections.Immutable;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
 
 namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
 
@@ -67,17 +68,120 @@ internal sealed class SubscriptionRegistry
     /// </summary>
     /// <param name="subscriptionId">The subscription identifier.</param>
     /// <param name="bindings">The tag bindings for the subscription.</param>
-    public void Register(long subscriptionId, IReadOnlyList<TagBinding> bindings)
+    public void Register(long subscriptionId, IReadOnlyList<TagBinding> bindings) =>
+        Bind(subscriptionId, bindings, requireExisting: false, expectedSession: null);
+
+    /// <summary>
+    ///     <see cref="Register"/>, but only if the gateway session is still
+    ///     <paramref name="expectedSession"/> — the session the SubscribeBulk that produced
+    ///     <paramref name="bindings"/> was issued on (#861). Item handles are per session and the next
+    ///     session reuses the numbers, so registering a reply that straddled a reconnect would bind
+    ///     dead handles that collide with live ones.
+    /// </summary>
+    /// <param name="subscriptionId">The subscription identifier.</param>
+    /// <param name="bindings">The tag bindings for the subscription.</param>
+    /// <param name="expectedSession">The <see cref="CurrentSession"/> captured before the SubscribeBulk.</param>
+    /// <returns>False (nothing registered) when the session changed in the meantime.</returns>
+    public bool TryRegister(long subscriptionId, IReadOnlyList<TagBinding> bindings, int expectedSession) =>
+        Bind(subscriptionId, bindings, requireExisting: false, expectedSession) == BindOutcome.Bound;
+
+    /// <summary>
+    ///     Replace an existing subscription's bindings with the item handles a post-reconnect
+    ///     SubscribeBulk returned, rebuilding the reverse fan-out map so events on the new
+    ///     handles dispatch and the now-dead pre-reconnect handles are dropped. No-op when the
+    ///     subscription id is unknown (it was unsubscribed during the reconnect window).
+    /// </summary>
+    /// <param name="subscriptionId">The subscription identifier.</param>
+    /// <param name="newBindings">The new tag bindings after reconnection.</param>
+    public void Rebind(long subscriptionId, IReadOnlyList<TagBinding> newBindings) =>
+        Bind(subscriptionId, newBindings, requireExisting: true, expectedSession: null);
+
+    /// <summary><see cref="Rebind"/>, but only while the session is still <paramref name="expectedSession"/> (#861).</summary>
+    /// <param name="subscriptionId">The subscription identifier.</param>
+    /// <param name="newBindings">The new tag bindings.</param>
+    /// <param name="expectedSession">The session the SubscribeBulk was issued on.</param>
+    /// <returns>
+    ///     <see cref="BindOutcome.Bound"/>; <see cref="BindOutcome.UnknownId"/> when the subscription was
+    ///     removed meanwhile (the caller must release the handles it just acquired — nothing will
+    ///     ever unsubscribe them, M-a); <see cref="BindOutcome.SessionChanged"/> when the session moved on.
+    /// </returns>
+    public BindOutcome TryRebind(long subscriptionId, IReadOnlyList<TagBinding> newBindings, int expectedSession) =>
+        Bind(subscriptionId, newBindings, requireExisting: true, expectedSession);
+
+    /// <summary>
+    ///     <see cref="Upsert"/> fenced by <paramref name="expectedSession"/> (the probe watcher, #859 I-A).
+    /// </summary>
+    /// <param name="subscriptionId">The subscription identifier.</param>
+    /// <param name="bindings">The tag bindings, all issued by <paramref name="expectedSession"/>.</param>
+    /// <param name="expectedSession">The session the bindings' handles belong to.</param>
+    /// <returns>False when the session changed (nothing published).</returns>
+    public bool TryUpsert(long subscriptionId, IReadOnlyList<TagBinding> bindings, int expectedSession) =>
+        Bind(subscriptionId, bindings, requireExisting: false, expectedSession) == BindOutcome.Bound;
+
+    /// <summary>Outcome of a fenced bind.</summary>
+    public enum BindOutcome
     {
-        var entry = new SubscriptionEntry(subscriptionId, bindings);
-        _bySubscriptionId[subscriptionId] = entry;
-        foreach (var binding in bindings)
+        /// <summary>Bound.</summary>
+        Bound,
+        /// <summary>The subscription id is no longer tracked (rebind only).</summary>
+        UnknownId,
+        /// <summary>The gateway session changed since the handles were issued.</summary>
+        SessionChanged,
+    }
+
+    private BindOutcome Bind(
+        long subscriptionId, IReadOnlyList<TagBinding> bindings, bool requireExisting, int? expectedSession)
+    {
+        bool promoted;
+        // The reverse-map writes and the early-event promotion are ONE critical section under the
+        // park gate (#859 review): a dispatch that resolves a handle through the map must never
+        // run between the write and the promotion, or a later live event would overtake the
+        // parked initial value. ResolveOrPark takes the same gate whenever anything is held. The
+        // session check is in here too, so it is atomic with StartNewSession's invalidation.
+        lock (_parkGate)
         {
-            if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected
-            _subscribersByItemHandle.AddOrUpdate(
-                binding.ItemHandle,
-                _ => [subscriptionId],
-                (_, set) => set.Add(subscriptionId));
+            if (expectedSession is { } session && session != _session) return BindOutcome.SessionChanged;
+
+            if (_bySubscriptionId.TryGetValue(subscriptionId, out var oldEntry))
+            {
+                // Drop this subscription from every reverse-map set it currently appears in. Only
+                // this id is removed, so another subscription that now owns the same handle number
+                // (a reused handle in a new session) keeps its entry.
+                RemoveFromReverseMap(oldEntry);
+            }
+            else if (requireExisting)
+            {
+                return BindOutcome.UnknownId;
+            }
+
+            _bySubscriptionId[subscriptionId] = new SubscriptionEntry(subscriptionId, bindings, _session);
+            foreach (var binding in bindings)
+            {
+                if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected
+                _subscribersByItemHandle.AddOrUpdate(
+                    binding.ItemHandle,
+                    _ => [subscriptionId],
+                    (_, set) => set.Add(subscriptionId));
+            }
+            AfterMapWriteForTests?.Invoke();
+            promoted = PromoteParkedEventsLocked(bindings);
+        }
+        if (promoted) ParkedEventsReady?.Invoke();
+        return BindOutcome.Bound;
+    }
+
+    private void RemoveFromReverseMap(SubscriptionEntry entry)
+    {
+        foreach (var binding in entry.Bindings)
+        {
+            if (binding.ItemHandle <= 0) continue;
+            if (!_subscribersByItemHandle.TryGetValue(binding.ItemHandle, out var set)) continue;
+            var remaining = set.Remove(entry.SubscriptionId);
+            if (remaining.IsEmpty)
+            {
+                _subscribersByItemHandle.TryRemove(binding.ItemHandle, out _);
+            }
+            else _subscribersByItemHandle[binding.ItemHandle] = remaining;
         }
     }
 
@@ -87,23 +191,27 @@ internal sealed class SubscriptionRegistry
     /// </summary>
     /// <param name="subscriptionId">The subscription identifier.</param>
     /// <returns>The bindings for the subscription, or null if not found.</returns>
-    public IReadOnlyList<TagBinding>? Remove(long subscriptionId)
+    public IReadOnlyList<TagBinding>? Remove(long subscriptionId) => Remove(subscriptionId, out _);
+
+    /// <summary>
+    ///     Remove a subscription, also reporting whether its handles belong to the CURRENT gateway
+    ///     session. A subscription not yet rebound after a reconnect still carries the previous
+    ///     session's handles, whose numbers the new session reuses — the caller must not send those
+    ///     to <c>UnsubscribeBulk</c>, or it would unadvise some other live item.
+    /// </summary>
+    /// <param name="subscriptionId">The subscription identifier.</param>
+    /// <param name="liveInCurrentSession">True when the bindings' handles were issued by the current session.</param>
+    /// <returns>The bindings for the subscription, or null if not found.</returns>
+    public IReadOnlyList<TagBinding>? Remove(long subscriptionId, out bool liveInCurrentSession)
     {
-        if (!_bySubscriptionId.TryRemove(subscriptionId, out var entry)) return null;
-
-        foreach (var binding in entry.Bindings)
+        lock (_parkGate)
         {
-            if (binding.ItemHandle <= 0) continue;
-            if (!_subscribersByItemHandle.TryGetValue(binding.ItemHandle, out var set)) continue;
-            var remaining = set.Remove(subscriptionId);
-            if (remaining.IsEmpty)
-            {
-                _subscribersByItemHandle.TryRemove(binding.ItemHandle, out _);
-            }
-            else _subscribersByItemHandle[binding.ItemHandle] = remaining;
+            liveInCurrentSession = false;
+            if (!_bySubscriptionId.TryRemove(subscriptionId, out var entry)) return null;
+            RemoveFromReverseMap(entry);
+            liveInCurrentSession = entry.Session == _session;
+            return entry.Bindings;
         }
-
-        return entry.Bindings;
     }
 
     /// <summary>
@@ -128,7 +236,6 @@ internal sealed class SubscriptionRegistry
         return result;
     }
 
-
     /// <summary>
     ///     Drop every subscription and the whole reverse map. Used by the driver's runtime teardown:
     ///     once the gateway session is closed every item handle in here is dead, and the host
@@ -136,8 +243,376 @@ internal sealed class SubscriptionRegistry
     /// </summary>
     public void Clear()
     {
-        _bySubscriptionId.Clear();
-        _subscribersByItemHandle.Clear();
+        lock (_parkGate)
+        {
+            _bySubscriptionId.Clear();
+            _subscribersByItemHandle.Clear();
+            DiscardAllHeldLocked();
+        }
+    }
+
+    // ===== Early-event window (#859) =====
+    //
+    // MXAccess delivers an initial value on Advise, and nothing orders that OnDataChange behind the
+    // SubscribeBulk reply the driver registers the new item handles from:
+    //   * mxaccessgw's worker completes the STA command through a TaskCompletionSource created with
+    //     RunContinuationsAsynchronously, so the reply is framed by a thread-pool continuation while
+    //     the STA loop is already pumping the OnDataChange that Advise queued; the event queue is
+    //     drained by a separate loop that frames events as soon as it is signalled
+    //     (mxaccessgw Worker/Sta/StaCommandDispatcher.cs, Worker/Ipc/WorkerPipeSession.cs
+    //     RunEventDrainLoopAsync). The writer's control-before-event priority only orders frames
+    //     that are queued at the same time.
+    //   * the reply travels back on the unary Invoke RPC and the event on the StreamEvents RPC —
+    //     separate HTTP/2 streams with no cross-stream ordering at the client;
+    //   * and even after the reply lands, the pump's dispatch loop runs concurrently with the code
+    //     that turns the reply into bindings.
+    // So an event whose handle is not yet registered is NOT necessarily "stale after unsubscribe".
+    // While at least one subscribe is in flight such events are parked here (bounded) instead of
+    // dropped; a registration that binds a parked handle promotes its events to a ready queue that
+    // the EventPump's dispatch loop replays — on that same single thread, BEFORE any later event for
+    // the handle — so per-handle value order is preserved.
+    //
+    // Unclaimed parked events are trimmed by WINDOW SEQUENCE, not only when every window has
+    // closed (#859 second review — concurrent reads keep some window open almost permanently):
+    // each window gets a monotonically increasing sequence when it opens, and each parked event is
+    // stamped with the newest sequence at park time. An event belonging to window W is generated
+    // after W opened, so its stamp is >= W's sequence; once every window with a sequence <= the
+    // stamp has closed, nothing can claim it and it is discarded.
+    //
+    // Item handles are only meaningful within one gateway SESSION, and a new session reuses small
+    // integers. StartNewSession (called on reconnect, before the replay re-subscribes) discards
+    // every held event, invalidates the reverse map (so a new-session event can never resolve to
+    // the tag that owned that number in the old session — it parks until the replay's Rebind),
+    // and fences: an event parked or replayed by an older session's pump is dropped, and a
+    // subscribe whose reply straddled the reconnect cannot register its dead handles (#861).
+
+    /// <summary>
+    ///     Upper bound on events held (parked + promoted-but-not-yet-replayed) across all in-flight
+    ///     subscribes. Sized to hold the initial values of a large bulk subscribe (MXAccess sends
+    ///     one per advised item); overflow is dropped and counted by the pump, i.e. it degrades to
+    ///     the pre-#859 behaviour.
+    /// </summary>
+    public const int MaxParkedEvents = 100_000;
+
+    private readonly Lock _parkGate = new();
+    private readonly Dictionary<int, Queue<ParkedEvent>> _parkedByItemHandle = new();
+    private readonly Queue<ParkedEvent> _parkOrder = new(); // every parked event, in park order (trim index)
+    private readonly Queue<MxEvent> _ready = new();
+    private readonly SortedSet<long> _openWindows = new();
+    private long _windowSequence;
+
+    // Parked + ready. Written only under _parkGate; read lock-free (Volatile) by the dispatch fast path.
+    private int _heldCount;
+
+    // Open windows. Written only under _parkGate; read lock-free by the dispatch fast path.
+    private int _pendingSubscribes;
+    private int _session;
+
+    /// <summary>
+    ///     Test seam (#859 review): invoked inside the registration critical section, between the
+    ///     reverse-map writes and the promotion of parked events.
+    /// </summary>
+    internal Action? AfterMapWriteForTests { get; set; }
+
+    /// <summary>
+    ///     Test seam (M-1): parks <paramref name="ev"/> unconditionally, producing the parked list for a
+    ///     resolved handle that the atomic registration otherwise makes impossible.
+    /// </summary>
+    /// <param name="ev">The event to park.</param>
+    internal void ForceParkForTests(MxEvent ev)
+    {
+        lock (_parkGate)
+        {
+            var entry = new ParkedEvent(_windowSequence, ev.ItemHandle, ev);
+            if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var queue))
+            {
+                queue = new Queue<ParkedEvent>();
+                _parkedByItemHandle[ev.ItemHandle] = queue;
+            }
+            queue.Enqueue(entry);
+            _parkOrder.Enqueue(entry);
+            Volatile.Write(ref _heldCount, _heldCount + 1);
+        }
+    }
+
+    /// <summary>
+    ///     Raised when a registration promoted parked events to the ready queue, so the
+    ///     <see cref="EventPump"/> can wake its dispatch loop even if no further event arrives.
+    ///     Raised after the park gate is released but still on the registering thread, which may hold
+    ///     the driver's <c>_registryGate</c> or the probe watcher's <c>_syncLock</c>: a handler must
+    ///     only signal (the pump's handler is a channel <c>TryWrite</c>, safe there because the
+    ///     channel keeps the default <c>AllowSynchronousContinuations = false</c>, so the dispatch
+    ///     loop never runs inline under those locks).
+    /// </summary>
+    public event Action? ParkedEventsReady;
+
+    /// <summary>Gets the number of subscribes currently inside their reply window.</summary>
+    public int PendingSubscribeCount => Volatile.Read(ref _pendingSubscribes);
+
+    /// <summary>Gets the number of events currently held (parked or awaiting replay).</summary>
+    public int ParkedEventCount => Volatile.Read(ref _heldCount);
+
+    /// <summary>Gets the current gateway-session number (bumped by <see cref="StartNewSession"/>).</summary>
+    public int CurrentSession => Volatile.Read(ref _session);
+
+    /// <summary>
+    ///     A new gateway session replaced the old one (reconnect). Discards every held event,
+    ///     invalidates the reverse map — subscriptions are KEPT, but their old-session handles no
+    ///     longer resolve, so new-session events park until the replay rebinds each subscription —
+    ///     and bumps the session number that fences old pumps and straddling subscribes. NOT for the
+    ///     same-session replay-gap re-snapshot, whose handles stay valid until rebound.
+    /// </summary>
+    /// <returns>The new session number.</returns>
+    public int StartNewSession()
+    {
+        lock (_parkGate)
+        {
+            _subscribersByItemHandle.Clear();
+            DiscardAllHeldLocked();
+            // Volatile: the dispatch fast path reads it lock-free so a dying pump of the old session
+            // can never resolve an old-session event through the map once the new session exists.
+            Volatile.Write(ref _session, _session + 1);
+            return _session;
+        }
+    }
+
+    /// <summary>
+    ///     Opens the early-event window for one subscribe. Call BEFORE issuing <c>SubscribeBulk</c>
+    ///     and dispose AFTER the returned handles are registered (or the attempt was abandoned) —
+    ///     not later: while a window is open, events for unknown handles are parked, not dropped.
+    /// </summary>
+    /// <returns>A token that closes the window when disposed (idempotent).</returns>
+    public IDisposable BeginPendingSubscribe()
+    {
+        long sequence;
+        lock (_parkGate)
+        {
+            sequence = ++_windowSequence;
+            _openWindows.Add(sequence);
+            Volatile.Write(ref _pendingSubscribes, _openWindows.Count);
+        }
+        return new PendingSubscribeScope(this, sequence);
+    }
+
+    private void EndPendingSubscribe(long sequence)
+    {
+        lock (_parkGate)
+        {
+            _openWindows.Remove(sequence);
+            Volatile.Write(ref _pendingSubscribes, _openWindows.Count);
+            TrimUnclaimableLocked();
+        }
+    }
+
+    /// <summary>
+    ///     Discard parked events no open window can claim: those stamped before the oldest open
+    ///     window's sequence (all of them once no window is open). Promoted (ready) events are kept —
+    ///     they belong to a registered handle and await replay. Amortised O(1) per parked event.
+    /// </summary>
+    private void TrimUnclaimableLocked()
+    {
+        var oldestOpen = _openWindows.Count == 0 ? long.MaxValue : _openWindows.Min;
+        while (_parkOrder.TryPeek(out var head) && head.Stamp < oldestOpen)
+        {
+            _parkOrder.Dequeue();
+            if (head.Gone) continue; // already promoted or purged
+            // Per-handle queues are in park order too, so a live head of the global order is the
+            // head of its handle's queue.
+            var queue = _parkedByItemHandle[head.ItemHandle];
+            queue.Dequeue();
+            head.Release();
+            if (queue.Count == 0) _parkedByItemHandle.Remove(head.ItemHandle);
+            Volatile.Write(ref _heldCount, _heldCount - 1);
+        }
+    }
+
+    private void DiscardAllHeldLocked()
+    {
+        foreach (var parked in _parkOrder) parked.Release();
+        _parkOrder.Clear();
+        _parkedByItemHandle.Clear();
+        _ready.Clear();
+        Volatile.Write(ref _heldCount, 0);
+    }
+
+    /// <summary>
+    ///     Drop parked events for handles that were just unsubscribed (#859 second review, I-3
+    ///     interim). A trailing event for a removed handle can be parked during another subscribe's
+    ///     window; if MXAccess reuses that handle number for the other subscribe's item, promotion
+    ///     would hand the old item's value to the new tag. Call when <c>UnsubscribeBulk</c> for
+    ///     <paramref name="itemHandles"/> completes. It cannot catch an event still in flight
+    ///     after that, nor tell a genuine early value for a reused number from a trailing one (it
+    ///     drops both — a timeout is safer than another tag's value); a complete fix needs a
+    ///     worker sequence on the SubscribeBulk reply (mxaccessgw, cross-repo).
+    /// </summary>
+    /// <param name="itemHandles">The handles whose unsubscribe completed.</param>
+    /// <param name="session">
+    ///     The session the handles (and the unsubscribe) belong to, captured BEFORE the RPC. A slow or
+    ///     failed unsubscribe can complete after a reconnect; purging then would delete the NEW
+    ///     session's parked initial values for reused numbers, so it is a no-op (#859 I-B).
+    /// </param>
+    public void PurgeParked(IReadOnlyList<int> itemHandles, int session)
+    {
+        if (Volatile.Read(ref _heldCount) == 0 || itemHandles.Count == 0) return;
+        lock (_parkGate)
+        {
+            if (session != _session) return;
+            foreach (var handle in itemHandles)
+            {
+                if (!_parkedByItemHandle.Remove(handle, out var queue)) continue;
+                foreach (var parked in queue) parked.Release();
+                Volatile.Write(ref _heldCount, _heldCount - queue.Count);
+            }
+        }
+    }
+
+    /// <summary>
+    ///     Resolve the subscribers for <paramref name="ev"/>'s item handle, or — when the handle is
+    ///     unknown while a subscribe is in flight — park the event for replay. Also hands back the
+    ///     promoted events that must be dispatched BEFORE this one (one gate acquisition per event
+    ///     while anything is held).
+    /// </summary>
+    /// <remarks>
+    ///     ⚠️ The lock-free fast path relies on a single-dispatcher invariant: only the ONE live
+    ///     <see cref="EventPump"/>'s dispatch thread ever parks (increments <c>_heldCount</c>), and it
+    ///     is the thread calling this. So when it reads <c>_heldCount == 0</c>, no older event for
+    ///     this handle can be waiting anywhere — it would have to have been parked by this same
+    ///     thread earlier — and resolving through the map cannot overtake one. Registrations only
+    ///     move events from parked to ready (count unchanged) and the dispatch thread only removes
+    ///     them. A second concurrently-dispatching pump on one registry would break this.
+    /// </remarks>
+    /// <param name="ev">The data-change event.</param>
+    /// <param name="session">The calling pump's session (<see cref="CurrentSession"/> at its creation).</param>
+    /// <param name="subscribers">The subscribers when the outcome is <see cref="EarlyEventOutcome.Resolved"/>.</param>
+    /// <param name="replayFirst">Promoted early events to dispatch before this one, in order.</param>
+    /// <returns>What happened to the event.</returns>
+    public EarlyEventOutcome ResolveOrPark(
+        MxEvent ev, int session,
+        out IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers,
+        out IReadOnlyList<MxEvent> replayFirst)
+    {
+        replayFirst = [];
+        if (Volatile.Read(ref _heldCount) == 0 && Volatile.Read(ref _session) == session)
+        {
+            subscribers = ResolveSubscribers(ev.ItemHandle);
+            if (subscribers.Count > 0) return EarlyEventOutcome.Resolved;
+            if (Volatile.Read(ref _pendingSubscribes) == 0) return EarlyEventOutcome.Stale;
+        }
+
+        lock (_parkGate)
+        {
+            if (session != _session)
+            {
+                subscribers = [];
+                return EarlyEventOutcome.Stale; // a dying pump of an earlier session
+            }
+
+            subscribers = ResolveSubscribers(ev.ItemHandle);
+            if (subscribers.Count > 0)
+            {
+                // Defensive (M-1): registration promotes atomically with its map writes, so a
+                // resolved handle should hold no parked list. If one ever does, promote it now
+                // (fail safe) — parking this LIVE event behind it would lose it to a later trim if
+                // no registration ever claimed the list.
+                if (_parkedByItemHandle.Remove(ev.ItemHandle, out var stranded))
+                {
+                    foreach (var parked in stranded)
+                    {
+                        _ready.Enqueue(parked.Release());
+                    }
+                }
+                replayFirst = TakeReadyLocked();
+                return EarlyEventOutcome.Resolved;
+            }
+
+            replayFirst = TakeReadyLocked();
+            if (_pendingSubscribes == 0) return EarlyEventOutcome.Stale;
+            if (_heldCount >= MaxParkedEvents) return EarlyEventOutcome.Overflow;
+
+            var entry = new ParkedEvent(_windowSequence, ev.ItemHandle, ev);
+            if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var queue))
+            {
+                queue = new Queue<ParkedEvent>();
+                _parkedByItemHandle[ev.ItemHandle] = queue;
+            }
+            queue.Enqueue(entry);
+            _parkOrder.Enqueue(entry);
+            Volatile.Write(ref _heldCount, _heldCount + 1);
+            return EarlyEventOutcome.Parked;
+        }
+    }
+
+    /// <summary>
+    ///     Dequeue every event a registration promoted, in arrival order. Called only from the
+    ///     EventPump's single dispatch thread (the wake path). A pump of an earlier session gets nothing.
+    /// </summary>
+    /// <param name="session">The calling pump's session.</param>
+    /// <returns>The promoted events; empty when none are waiting.</returns>
+    public IReadOnlyList<MxEvent> TakeReadyEvents(int session)
+    {
+        if (Volatile.Read(ref _heldCount) == 0) return [];
+        lock (_parkGate)
+        {
+            return session != _session ? [] : TakeReadyLocked();
+        }
+    }
+
+    private IReadOnlyList<MxEvent> TakeReadyLocked()
+    {
+        if (_ready.Count == 0) return [];
+        var taken = _ready.ToArray();
+        _ready.Clear();
+        Volatile.Write(ref _heldCount, _heldCount - taken.Length);
+        return taken;
+    }
+
+    private bool PromoteParkedEventsLocked(IReadOnlyList<TagBinding> bindings)
+    {
+        if (_parkedByItemHandle.Count == 0) return false;
+        var promoted = false;
+        foreach (var binding in bindings)
+        {
+            if (binding.ItemHandle <= 0) continue;
+            if (!_parkedByItemHandle.Remove(binding.ItemHandle, out var queue)) continue;
+            foreach (var parked in queue)
+            {
+                _ready.Enqueue(parked.Release()); // stays counted in _heldCount until replayed
+            }
+            promoted = true;
+        }
+        return promoted;
+    }
+
+    private sealed class ParkedEvent(long stamp, int itemHandle, MxEvent ev)
+    {
+        /// <summary>Gets the newest window sequence when the event was parked.</summary>
+        public long Stamp { get; } = stamp;
+        /// <summary>Gets the event's item handle.</summary>
+        public int ItemHandle { get; } = itemHandle;
+        /// <summary>Gets the event; null once it has left the parked set (M-c — not pinned by the trim index).</summary>
+        public MxEvent? Event { get; private set; } = ev;
+        /// <summary>Gets whether the event has left the parked set (promoted, purged or trimmed).</summary>
+        public bool Gone => Event is null;
+
+        /// <summary>Takes the event out of this entry, so the trim index no longer pins it.</summary>
+        /// <returns>The event.</returns>
+        public MxEvent Release()
+        {
+            var ev = Event!;
+            Event = null;
+            return ev;
+        }
+    }
+
+    private sealed class PendingSubscribeScope(SubscriptionRegistry owner, long sequence) : IDisposable
+    {
+        private int _disposed;
+
+        /// <summary>Closes this subscribe's early-event window (once).</summary>
+        public void Dispose()
+        {
+            if (Interlocked.Exchange(ref _disposed, 1) == 0) owner.EndPendingSubscribe(sequence);
+        }
     }
 
     /// <summary>Snapshot every active binding for diagnostic output.</summary>
@@ -155,41 +630,14 @@ internal sealed class SubscriptionRegistry
         [.. _bySubscriptionId.Values.Select(entry => (entry.SubscriptionId, entry.Bindings))];
 
     /// <summary>
-    ///     Replace an existing subscription's bindings with the item handles a post-reconnect
-    ///     SubscribeBulk returned, rebuilding the reverse fan-out map so events on the new
-    ///     handles dispatch and the now-dead pre-reconnect handles are dropped. No-op when the
-    ///     subscription id is unknown (it was unsubscribed during the reconnect window).
+    ///     <see cref="SnapshotEntries"/> limited to subscriptions whose handles were NOT issued by
+    ///     <paramref name="session"/> — the reconnect replay's work list. A subscription registered after
+    ///     the reconnect already holds live handles; re-subscribing it would double-advise (#859 I-C).
     /// </summary>
-    /// <param name="subscriptionId">The subscription identifier.</param>
-    /// <param name="newBindings">The new tag bindings after reconnection.</param>
-    public void Rebind(long subscriptionId, IReadOnlyList<TagBinding> newBindings)
-    {
-        if (!_bySubscriptionId.TryGetValue(subscriptionId, out var oldEntry)) return;
-
-        // Drop this subscription from every reverse-map set it currently appears in. The
-        // pre-reconnect item handles are stale once the gw re-issues fresh ones.
-        foreach (var binding in oldEntry.Bindings)
-        {
-            if (binding.ItemHandle <= 0) continue;
-            if (!_subscribersByItemHandle.TryGetValue(binding.ItemHandle, out var set)) continue;
-            var remaining = set.Remove(subscriptionId);
-            if (remaining.IsEmpty)
-            {
-                _subscribersByItemHandle.TryRemove(binding.ItemHandle, out _);
-            }
-            else _subscribersByItemHandle[binding.ItemHandle] = remaining;
-        }
-
-        _bySubscriptionId[subscriptionId] = new SubscriptionEntry(subscriptionId, newBindings);
-        foreach (var binding in newBindings)
-        {
-            if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected
-            _subscribersByItemHandle.AddOrUpdate(
-                binding.ItemHandle,
-                _ => [subscriptionId],
-                (_, set) => set.Add(subscriptionId));
-        }
-    }
+    /// <param name="session">The current session.</param>
+    /// <returns>The stale subscriptions with their bindings.</returns>
+    public IReadOnlyList<(long SubscriptionId, IReadOnlyList<TagBinding> Bindings)> SnapshotEntriesNotIn(int session) =>
+        [.. _bySubscriptionId.Values.Where(e => e.Session != session).Select(e => (e.SubscriptionId, e.Bindings))];
 
     /// <summary>
     ///     Per-subscription bookkeeping. <see cref="FullRefByItemHandle"/> is an index
@@ -206,14 +654,18 @@ internal sealed class SubscriptionRegistry
         public IReadOnlyList<TagBinding> Bindings { get; }
         /// <summary>Gets the index of full references by item handle.</summary>
         public IReadOnlyDictionary<int, string> FullRefByItemHandle { get; }
+        /// <summary>Gets the gateway session whose handles <see cref="Bindings"/> carries.</summary>
+        public int Session { get; }
 
         /// <summary>Initializes a new subscription entry.</summary>
         /// <param name="subscriptionId">The subscription identifier.</param>
         /// <param name="bindings">The tag bindings for the subscription.</param>
-        public SubscriptionEntry(long subscriptionId, IReadOnlyList<TagBinding> bindings)
+        /// <param name="session">The session the bindings' handles were issued by.</param>
+        public SubscriptionEntry(long subscriptionId, IReadOnlyList<TagBinding> bindings, int session)
         {
             SubscriptionId = subscriptionId;
             Bindings = bindings;
+            Session = session;
             var index = new Dictionary<int, string>(bindings.Count);
             foreach (var binding in bindings)
             {
@@ -227,6 +679,22 @@ internal sealed class SubscriptionRegistry
     }
 }
 
+/// <summary>Outcome of <see cref="SubscriptionRegistry.ResolveOrPark"/>.</summary>
+internal enum EarlyEventOutcome
+{
+    /// <summary>The handle is registered; dispatch to the returned subscribers.</summary>
+    Resolved,
+
+    /// <summary>No subscribe in flight — a genuinely stale event (after unsubscribe); drop it.</summary>
+    Stale,
+
+    /// <summary>Parked until the in-flight subscribe registers the handle.</summary>
+    Parked,
+
+    /// <summary>Would have been parked but <see cref="SubscriptionRegistry.MaxParkedEvents"/> was reached; dropped.</summary>
+    Overflow,
+}
+
 /// <summary>
 ///     One (full reference, gw item handle) pair returned by SubscribeBulk. Item handle is
 ///     zero or negative when the gateway rejected this individual tag (bad name, duplicate);