Merge #864/#865 Galaxy host status: per-template platform seed, event stream started before probe registration; registry and reconnect races fixed

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
2026-11-11 13:22:33 -05:00
parent 50b7b030a9 59d33ab938
commit b5d4e25905
12 changed files with 1197 additions and 89 deletions
@@ -15,33 +15,99 @@ public sealed class GatewayGalaxyPlatformSourceTests
        return o;
    }

-    /// <summary>Every request carries the platform template filter, no attributes, and the page size; replies are re-filtered.</summary>
-    [Fact]
-    public async Task Requests_filter_on_platform_templates_without_attributes_and_replies_are_refiltered()
+    /// <summary>
+    ///     A fake gateway that applies the request filters with the SAME semantics as the shared
+    ///     <c>GalaxyHierarchyProjector.MatchesFilters</c>: every <c>template_chain_contains</c> token must
+    ///     match (AND, case-insensitive substring over the chain), <c>category_ids</c> is an OR set. Pages
+    ///     the filtered result at the request's page size.
+    /// </summary>
+    private sealed class FilteringGateway(params GalaxyObject[] galaxy)
    {
-        var requests = new List<DiscoverHierarchyRequest>();
-        Task<DiscoverHierarchyReply> Discover(DiscoverHierarchyRequest req, CancellationToken ct)
+        public List<DiscoverHierarchyRequest> Requests { get; } = [];
+
+        public Task<DiscoverHierarchyReply> Discover(DiscoverHierarchyRequest req, CancellationToken ct)
        {
-            requests.Add(req);
+            Requests.Add(req.Clone());
+            var matched = galaxy.Where(o => Matches(o, req)).ToList();
+            var offset = req.PageToken == "" ? 0 : int.Parse(req.PageToken.Split(':')[1]);
            var reply = new DiscoverHierarchyReply();
-            if (req.PageToken == "")
+            reply.Objects.Add(matched.Skip(offset).Take(req.PageSize));
+            if (offset + req.PageSize < matched.Count)
            {
-                reply.Objects.Add([Obj("Plat01", "$WinPlatform"), Obj("Tank01", "$UserDefined")]);
-                reply.NextPageToken = "p2";
-            }
-            else
-            {
-                reply.Objects.Add([Obj("Eng01", "$AppEngine", "$WinPlatform"), Obj("Plat01", "$WinPlatform"), Obj("", "$WinPlatform")]);
+                reply.NextPageToken = $"{string.Join(",", req.TemplateChainContains)}:{offset + req.PageSize}";
            }
            return Task.FromResult(reply);
        }

-        var names = await GatewayGalaxyPlatformSource.QueryAsync(Discover, pageSize: 50, CancellationToken.None);
+        private static bool Matches(GalaxyObject obj, DiscoverHierarchyRequest req)
+        {
+            if (req.CategoryIds.Count > 0 && !req.CategoryIds.Contains(obj.CategoryId)) return false;
+            foreach (var filter in req.TemplateChainContains)
+            {
+                if (!obj.TemplateChain.Any(t => t.Contains(filter, StringComparison.OrdinalIgnoreCase))) return false;
+            }
+            return true;
+        }
+    }

-        names.ShouldBe(["Plat01", "Eng01"]);
-        requests.Count.ShouldBe(2);
-        requests.ShouldAllBe(r => r.PageSize == 50 && r.IncludeAttributes == false);
-        requests.ShouldAllBe(r => r.TemplateChainContains.Contains("$WinPlatform") && r.TemplateChainContains.Contains("$AppEngine"));
+    /// <summary>
+    ///     #864: against a gateway that ANDs template filters (as the real one does), every deployed
+    ///     <c>$WinPlatform</c> AND every <c>$AppEngine</c> is found — the old single request carrying both
+    ///     tokens matched none. Shape of the VD03 DEV galaxy: 3 platforms, 2 engines.
+    /// </summary>
+    [Fact]
+    public async Task Finds_every_platform_and_engine_against_an_AND_filtering_gateway()
+    {
+        var gw = new FilteringGateway(
+            Obj("CTRX_DEV", "$WinPlatform"),
+            Obj("DevWinPlatform", "$WinPlatform"),
+            Obj("JoePlatform", "$JoePlatformTemplate", "$WinPlatform"),
+            Obj("DevAppEngine", "$AppEngine"),
+            Obj("JoeAppEngine", "$AppEngine"),
+            Obj("Tank01", "$UserDefined"),
+            Obj("Area01", "$Area"));
+
+        var names = await GatewayGalaxyPlatformSource.QueryAsync(gw.Discover, pageSize: 50, CancellationToken.None);
+
+        names.ShouldBe(["CTRX_DEV", "DevWinPlatform", "JoePlatform", "DevAppEngine", "JoeAppEngine"]);
+        gw.Requests.ShouldAllBe(r => r.TemplateChainContains.Count == 1 && r.PageSize == 50 && r.IncludeAttributes == false);
+        gw.Requests.Select(r => r.TemplateChainContains[0]).Distinct().ShouldBe(["$WinPlatform", "$AppEngine"]);
+    }
+
+    /// <summary>Each per-template sweep pages to completion independently; tag names are de-duplicated across sweeps.</summary>
+    [Fact]
+    public async Task Each_template_sweep_pages_independently_and_names_are_deduplicated()
+    {
+        var gw = new FilteringGateway(
+            Obj("Plat01", "$WinPlatform"),
+            Obj("Plat02", "$WinPlatform"),
+            Obj("Plat03", "$WinPlatform"),
+            // An object whose chain carries both templates is reported once.
+            Obj("Hybrid01", "$AppEngine", "$WinPlatform"),
+            Obj("Eng01", "$AppEngine"),
+            Obj("Eng02", "$AppEngine"),
+            Obj("", "$WinPlatform"));
+
+        var names = await GatewayGalaxyPlatformSource.QueryAsync(gw.Discover, pageSize: 2, CancellationToken.None);
+
+        names.ShouldBe(["Plat01", "Plat02", "Plat03", "Hybrid01", "Eng01", "Eng02"]);
+        // $WinPlatform: 5 matches (incl. the empty tag) => 3 pages; $AppEngine: 3 matches => 2 pages.
+        gw.Requests.Count(r => r.TemplateChainContains[0] == "$WinPlatform").ShouldBe(3);
+        gw.Requests.Count(r => r.TemplateChainContains[0] == "$AppEngine").ShouldBe(2);
+    }
+
+    /// <summary>A gateway that ignores the filter still yields only platforms (replies are re-filtered on the chain).</summary>
+    [Fact]
+    public async Task Replies_are_refiltered_on_the_template_chain()
+    {
+        Task<DiscoverHierarchyReply> Discover(DiscoverHierarchyRequest req, CancellationToken ct)
+        {
+            var reply = new DiscoverHierarchyReply();
+            reply.Objects.Add([Obj("Plat01", "$WinPlatform"), Obj("Tank01", "$UserDefined"), Obj("Eng01", "$AppEngine")]);
+            return Task.FromResult(reply);
+        }
+
+        (await GatewayGalaxyPlatformSource.QueryAsync(Discover, pageSize: 50, CancellationToken.None)).ShouldBe(["Plat01", "Eng01"]);
    }

    /// <summary>A gateway that strips the template chain is trusted to have filtered server-side.</summary>
@@ -844,6 +844,7 @@ public sealed class EarlyEventWindowTests
            "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber)
        {
            RecreateOverrideForTests = _ => Task.CompletedTask,
+            StreamRequestWait = TimeSpan.FromMilliseconds(100), // advises in the reopen window on purpose (#865 bound)
        };
        var delivered = new ConcurrentQueue<(string Ref, double Value)>();
        driver.OnDataChange += (_, e) => delivered.Enqueue((e.FullReference, (double)e.Snapshot.Value!));
@@ -897,6 +898,31 @@ public sealed class EarlyEventWindowTests
        driver.SubscriptionsForTests.ResolveSubscribers(77).ShouldBeEmpty();
    }

+    /// <summary>
+    ///     #865 stress finding: the dispatch fast path looked the handle up, then checked for an open window —
+    ///     not atomically. A subscribe that registered the handle AND closed its window in between made the
+    ///     event "stale" and its initial value was dropped for good. It must resolve.
+    /// </summary>
+    [Fact]
+    public void A_registration_landing_between_the_fast_path_lookup_and_the_window_check_still_resolves()
+    {
+        var registry = new SubscriptionRegistry();
+        var session = registry.CurrentSession;
+        var window = registry.BeginPendingSubscribe();
+        registry.AfterFastPathMissForTests = () =>
+        {
+            registry.AfterFastPathMissForTests = null;
+            registry.TryRegister(registry.NextSubscriptionId(), [new TagBinding("Tank.A", 7)], session).ShouldBeTrue();
+            window.Dispose(); // the subscribe's window closes right after its registration
+        };
+
+        registry.ResolveOrPark(DataChange(7, 1), session, out var subscribers, out _).ShouldBe(EarlyEventOutcome.Resolved);
+        subscribers.Single().FullReference.ShouldBe("Tank.A");
+
+        // Positive control for the Stale branch: an unknown handle with no window open is still dropped.
+        registry.ResolveOrPark(DataChange(8, 1), session, out _, out _).ShouldBe(EarlyEventOutcome.Stale);
+    }
+
    /// <summary>I-B: a purge that completes after a reconnect does not delete the new session's parked values.</summary>
    [Fact]
    public void A_purge_for_an_older_session_is_a_no_op()
@@ -1027,6 +1053,9 @@ public sealed class EarlyEventWindowTests
            "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber)
        {
            RecreateOverrideForTests = _ => Task.CompletedTask,
+            // These tests advise in the reopen → replay window on purpose (fencing, not values). #865 makes
+            // such an advise wait for the replay's pump, bounded; keep the bound short so they don't idle.
+            StreamRequestWait = TimeSpan.FromMilliseconds(100),
        };
        await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
        return driver;
@@ -0,0 +1,62 @@
+using System.Runtime.CompilerServices;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
+
+/// <summary>
+///     #865: <see cref="EventPump.StreamRequested"/> — what advise paths wait on so an initial value is
+///     never pushed before the stream is requested. It must complete once the RPC is started (before any
+///     event), and must never strand a waiter on a pump whose loop died before getting that far.
+/// </summary>
+public sealed class EventPumpStreamRequestedTests
+{
+    private sealed class ScriptedSubscriber(Func<CancellationToken, IAsyncEnumerable<MxEvent>> stream) : IGalaxySubscriber
+    {
+        public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
+            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
+            => Task.FromResult<IReadOnlyList<SubscribeResult>>([]);
+
+        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;
+
+        public IAsyncEnumerable<MxEvent> StreamEventsAsync(CancellationToken cancellationToken) => stream(cancellationToken);
+    }
+
+    /// <summary>A stream that is requested and then never yields: StreamRequested completes; the RPC started first.</summary>
+    [Fact]
+    public async Task Completes_once_the_stream_is_requested_before_any_event()
+    {
+        var rpcStarted = 0;
+        async IAsyncEnumerable<MxEvent> Never([EnumeratorCancellation] CancellationToken ct)
+        {
+            Interlocked.Exchange(ref rpcStarted, 1); // the synchronous first step — where a real call is created
+            await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
+            yield break;
+        }
+
+        await using var pump = new EventPump(new ScriptedSubscriber(Never), new SubscriptionRegistry());
+        pump.StreamRequested.IsCompleted.ShouldBeFalse("not before Start");
+        pump.Start();
+
+        await TestWait.ForAsync(pump.StreamRequested, "StreamRequested must complete once the stream call is started");
+        Volatile.Read(ref rpcStarted).ShouldBe(1, "StreamRequested must not precede the stream call");
+    }
+
+    /// <summary>A subscriber that throws before reaching the RPC still completes StreamRequested (no stranded waiter).</summary>
+    [Fact]
+    public async Task Completes_when_the_stream_fails_before_the_rpc()
+    {
+        var faulted = new TaskCompletionSource<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
+        await using var pump = new EventPump(
+            new ScriptedSubscriber(_ => throw new InvalidOperationException("not connected")),
+            new SubscriptionRegistry(),
+            onStreamFault: ex => faulted.TrySetResult(ex));
+        pump.Start();
+
+        await TestWait.ForAsync(pump.StreamRequested, "a dead pump must not strand a StreamRequested waiter");
+        (await TestWait.ForAsync(faulted.Task, "positive control: the failure is still reported as a stream fault"))
+            .ShouldBeOfType<InvalidOperationException>();
+    }
+}
@@ -272,7 +272,10 @@ public sealed class GalaxyDriverFixRound1Tests
    [Fact]
    public async Task Production_path_queries_the_platform_source_at_init()
    {
-        var platforms = new ScriptedPlatformSource();
+        // An empty galaxy: since #865 a probe advise starts the event stream first, and this owned runtime's
+        // gRPC channel points at no real gateway — its stream would fault and (correctly) degrade health.
+        // The query itself is what this test pins.
+        var platforms = new ScriptedPlatformSource { Script = _ => [] };
        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, platformSource: platforms)
        {
            SessionOpenOverrideForTests = _ => Task.CompletedTask,
@@ -303,6 +306,7 @@ public sealed class GalaxyDriverFixRound1Tests

        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
        driver.HasEventPumpForTests.ShouldBeTrue("positive control: subscribe works on a live runtime");
+        var oldPump = driver.EventPumpInstanceForTests;

        sub.ParkUnsubscribe = true;
        var shutdown = driver.ShutdownAsync(CancellationToken.None);
@@ -330,7 +334,12 @@ public sealed class GalaxyDriverFixRound1Tests
        driver.TrackedSubscriptionCountForTests.ShouldBe(0);

        await driver.InitializeAsync("{}", CancellationToken.None);
-        driver.HasEventPumpForTests.ShouldBeFalse("the next runtime starts with no inherited pump");
+        // #865: the new runtime's background probe seed starts a pump of its own (its advises need a live
+        // stream) — but it must be a FRESH one, never the dead subscriber's.
+        await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1,
+            "positive control: the new runtime's background seed must advise its platform");
+        driver.EventPumpInstanceForTests.ShouldNotBeNull("the seed's probe advise must have started a pump")
+            .ShouldNotBeSameAs(oldPump, "the next runtime must not inherit the old pump");
    }

    // ───────────────────────────── IMPORTANT 3: bounded gateway close ─────────────────────────────
@@ -0,0 +1,419 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using Google.Protobuf.WellKnownTypes;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
+
+/// <summary>
+///     #865: the platform ScanState probes' initial value. The gateway pushes an advise's current value
+///     once, as its first event, and a FRESH <c>StreamEvents</c> (<c>after_worker_sequence = 0</c>) never
+///     replays — so a probe advised before the driver's event stream is attached never gets a value, and
+///     ScanState rarely changes again. The event pump used to start only on the first Read/Subscribe,
+///     so a driver with no authored tags (or any driver before its first read) had no host statuses.
+/// </summary>
+public sealed class ProbeInitialValueTests
+{
+    private static readonly string[] Platforms = ["CTRX_DEV", "DevWinPlatform", "DevAppEngine"];
+
+    private static GalaxyDriverOptions Opts() => new(
+        new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
+        new GalaxyMxAccessOptions("ProbeInitialValue"),
+        new GalaxyRepositoryOptions(WatchDeployEvents: false),
+        new GalaxyReconnectOptions());
+
+    private sealed class FixedPlatformSource(params string[] platforms) : IGalaxyPlatformSource
+    {
+        public Task<IReadOnlyList<string>> GetPlatformTagNamesAsync(CancellationToken cancellationToken)
+            => Task.FromResult<IReadOnlyList<string>>(platforms);
+    }
+
+    /// <summary>
+    ///     Models the gateway's event semantics: every advise pushes its current value exactly once, and
+    ///     only to a stream that is attached at that moment (no replay for a fresh stream). Records the
+    ///     order in which the stream was requested and the advises were made.
+    /// </summary>
+    private sealed class GatewayLikeSubscriber : IGalaxySubscriber
+    {
+        private readonly Lock _gate = new();
+        private Channel<MxEvent>? _attached;
+        private int _nextHandle;
+
+        /// <summary>"stream" per StreamEvents call, "subscribe:&lt;refs&gt;" per SubscribeBulk, in call order.</summary>
+        public List<string> Calls { get; } = [];
+
+        /// <summary>Initial values pushed while no stream was attached — lost, as on the real gateway.</summary>
+        public int DroppedInitialValues;
+
+        private int _streams;
+
+        /// <summary>Streams beyond this count end immediately after attaching (a gateway refusing/ending them).</summary>
+        public int StreamsThatStayOpen = int.MaxValue;
+
+        /// <summary>The ScanState value the next advise reports.</summary>
+        public volatile bool ScanStateRunning = true;
+
+        public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
+            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
+        {
+            var results = new List<SubscribeResult>();
+            lock (_gate)
+            {
+                Calls.Add("subscribe:" + string.Join(",", fullReferences));
+                foreach (var r in fullReferences)
+                {
+                    var h = ++_nextHandle;
+                    results.Add(new SubscribeResult { TagAddress = r, ItemHandle = h, WasSuccessful = true });
+                    var initial = new MxEvent
+                    {
+                        Family = MxEventFamily.OnDataChange,
+                        ItemHandle = h,
+                        Value = new MxValue { BoolValue = ScanStateRunning },
+                        Quality = 192,
+                        SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+                    };
+                    if (_attached is null || !_attached.Writer.TryWrite(initial)) Interlocked.Increment(ref DroppedInitialValues);
+                }
+            }
+            return Task.FromResult<IReadOnlyList<SubscribeResult>>(results);
+        }
+
+        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;
+
+        public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            // Attached synchronously on the first MoveNext — the point a real gRPC call is issued.
+            var mine = Channel.CreateUnbounded<MxEvent>();
+            lock (_gate)
+            {
+                Calls.Add("stream");
+                _attached = mine;
+            }
+            if (Interlocked.Increment(ref _streams) > StreamsThatStayOpen)
+            {
+                lock (_gate) { if (ReferenceEquals(_attached, mine)) _attached = null; }
+                yield break; // the gateway completed this stream at once
+            }
+            try
+            {
+                await foreach (var ev in mine.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+                {
+                    yield return ev;
+                }
+            }
+            finally
+            {
+                lock (_gate)
+                {
+                    if (ReferenceEquals(_attached, mine)) _attached = null;
+                }
+            }
+        }
+
+        /// <summary>Fault the currently attached stream (the gateway dropping it).</summary>
+        public void FaultAttachedStream(Exception cause)
+        {
+            lock (_gate) _attached.ShouldNotBeNull("a stream must be attached").Writer.TryComplete(cause);
+        }
+
+        public string[] CallsSnapshot()
+        {
+            lock (_gate) return [.. Calls];
+        }
+    }
+
+    private static HostState? StateOf(GalaxyDriver driver, string host) =>
+        driver.GetHostStatuses().FirstOrDefault(s => s.HostName == host)?.State;
+
+    private static string Describe(GalaxyDriver driver, GatewayLikeSubscriber sub) =>
+        $"hosts=[{string.Join(", ", driver.GetHostStatuses().Select(s => $"{s.HostName}={s.State}"))}] " +
+        $"calls=[{string.Join(" | ", sub.CallsSnapshot())}] dropped={sub.DroppedInitialValues} pump={driver.HasEventPumpForTests}";
+
+    /// <summary>
+    ///     A driver with NO authored tags — nothing ever Reads or Subscribes — still reports every seeded
+    ///     platform with its initial ScanState. Before #865 the pump never started, every initial value
+    ///     was pushed to no stream, and only the transport entry ever appeared.
+    /// </summary>
+    [Fact]
+    public async Task Driver_with_no_authored_tags_reports_every_platform_from_its_initial_ScanState()
+    {
+        var sub = new GatewayLikeSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub,
+            platformSource: new FixedPlatformSource(Platforms));
+
+        await driver.InitializeAsync("{}", CancellationToken.None);
+
+        await TestWait.UntilAsync(
+            () => Platforms.All(p => StateOf(driver, p) == HostState.Running),
+            "every seeded platform must surface Running from its initial ScanState, with no Read/Subscribe",
+            () => Describe(driver, sub));
+        sub.DroppedInitialValues.ShouldBe(0, "no probe's initial value may be pushed before the stream is attached");
+        driver.TrackedSubscriptionCountForTests.ShouldBe(1, "only the probe entry — no data subscription was ever made");
+    }
+
+    /// <summary>
+    ///     Ordering: the event stream is requested BEFORE the first probe SubscribeBulk. On the old code
+    ///     the probe advise went out with no stream at all.
+    /// </summary>
+    [Fact]
+    public async Task Probe_subscribe_is_issued_only_after_the_event_stream_is_requested()
+    {
+        var sub = new GatewayLikeSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub,
+            platformSource: new FixedPlatformSource(Platforms));
+
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        await TestWait.UntilAsync(
+            () => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == Platforms.Length,
+            "positive control: the background seed must advise every platform",
+            () => Describe(driver, sub));
+
+        var calls = sub.CallsSnapshot();
+        var firstSubscribe = Array.FindIndex(calls, c => c.StartsWith("subscribe:", StringComparison.Ordinal));
+        var firstStream = Array.IndexOf(calls, "stream");
+        firstSubscribe.ShouldBeGreaterThanOrEqualTo(0);
+        firstStream.ShouldBeGreaterThanOrEqualTo(0, $"the probe seed must start the event stream; calls=[{string.Join(" | ", calls)}]");
+        firstStream.ShouldBeLessThan(firstSubscribe, $"the stream must be requested before the probe advise; calls=[{string.Join(" | ", calls)}]");
+    }
+
+    /// <summary>
+    ///     A reconnect replay re-advises the probes on the recreated session; their new initial values must
+    ///     arrive too (the replay restarts the pump first). The value flips to Stopped so the presence wait
+    ///     can only pass on a post-replay event.
+    /// </summary>
+    [Fact]
+    public async Task Reconnect_replay_delivers_the_re_advised_probes_initial_values()
+    {
+        var sub = new GatewayLikeSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub,
+            platformSource: new FixedPlatformSource(Platforms));
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        await TestWait.UntilAsync(
+            () => Platforms.All(p => StateOf(driver, p) == HostState.Running),
+            "positive control: the initial seed reports Running",
+            () => Describe(driver, sub));
+
+        sub.ScanStateRunning = false;
+        await driver.InvokeReplayForTestAsync(CancellationToken.None);
+
+        await TestWait.UntilAsync(
+            () => Platforms.All(p => StateOf(driver, p) == HostState.Stopped),
+            "the re-advised probes' initial values (Stopped) must reach the host statuses after a replay",
+            () => Describe(driver, sub));
+        sub.DroppedInitialValues.ShouldBe(0);
+    }
+
+    /// <summary>
+    ///     Review finding: a probe advised in the reopen → replay window (the session was swapped, the old
+    ///     pump still streams the dead session) must still get its initial value. The advise waits for the
+    ///     replay's restarted pump instead of advising onto the stale one (its value would route under the
+    ///     old session and be lost, and the replay skips a current-session handle) or starting a competing
+    ///     stream (the gateway's default single-subscriber mode refuses a second attach).
+    /// </summary>
+    [Fact]
+    public async Task Probe_advised_between_reopen_and_replay_still_gets_its_initial_value()
+    {
+        var sub = new GatewayLikeSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub,
+            platformSource: new FixedPlatformSource(Platforms))
+        {
+            RecreateOverrideForTests = _ => Task.CompletedTask,
+            // Generous: the window advise must still be waiting when the replay below starts the pump, even
+            // on a stalled agent; the wait ends the moment that pump exists.
+            StreamRequestWait = TimeSpan.FromSeconds(20),
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        await TestWait.UntilAsync(
+            () => Platforms.All(p => StateOf(driver, p) == HostState.Running),
+            "positive control: the initial seed reports Running",
+            () => Describe(driver, sub));
+        var oldPump = driver.EventPumpInstanceForTests;
+
+        await driver.InvokeReopenForTestAsync(CancellationToken.None); // session swapped; no replay yet
+        sub.ScanStateRunning = false;
+        var callsBeforeWindow = sub.CallsSnapshot().Length;
+        // The window advise must WAIT for the replay's current-session pump rather than advise onto the
+        // stale one (or start a competing stream).
+        var sync = driver.ProbeWatcherForTests.ShouldNotBeNull()
+            .SyncPlatformsAsync([.. Platforms, "LatePlatform"], CancellationToken.None);
+
+        await driver.InvokeReplayForTestAsync(CancellationToken.None);
+        await TestWait.ForAsync(sync, "the window advise completes once the replay restarted the pump");
+
+        await TestWait.UntilAsync(
+            () => StateOf(driver, "LatePlatform") == HostState.Stopped
+                  && Platforms.All(p => StateOf(driver, p) == HostState.Stopped),
+            "the window advise and the replay's re-advises all get their initial values",
+            () => Describe(driver, sub));
+        driver.EventPumpInstanceForTests.ShouldNotBeSameAs(oldPump, "the replay restarted the stale pump");
+        var windowCalls = sub.CallsSnapshot()[callsBeforeWindow..];
+        var lateAdvise = Array.FindIndex(windowCalls, c => c.Contains("LatePlatform", StringComparison.Ordinal));
+        var newStream = Array.IndexOf(windowCalls, "stream");
+        newStream.ShouldBeGreaterThanOrEqualTo(0, $"the replay must start a new stream; calls=[{string.Join(" | ", windowCalls)}]");
+        lateAdvise.ShouldBeGreaterThan(newStream, $"the window advise must follow the new stream; calls=[{string.Join(" | ", windowCalls)}]");
+        windowCalls.Count(c => c == "stream").ShouldBe(1, "no competing stream may be started for the window advise");
+        sub.DroppedInitialValues.ShouldBe(0);
+    }
+
+    /// <summary>
+    ///     The data paths too: a first SubscribeAsync and a first ReadAsync on a driver whose pump was never
+    ///     started request the stream BEFORE advising, so their initial values arrive (a read of a value that
+    ///     never changes again would otherwise time out).
+    /// </summary>
+    [Fact]
+    public async Task First_subscribe_and_read_advise_only_after_the_stream_is_requested()
+    {
+        var sub = new GatewayLikeSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub)
+        {
+            ReadViaSubscribeDeadline = TimeSpan.FromSeconds(10),
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        var delivered = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
+        driver.OnDataChange += (_, e) => delivered.TrySetResult(e);
+
+        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
+        var calls = sub.CallsSnapshot();
+        Array.IndexOf(calls, "stream").ShouldBe(0, $"the stream must be requested before the advise; calls=[{string.Join(" | ", calls)}]");
+        (await TestWait.ForAsync(delivered.Task, "the subscribe's initial value must arrive", () => Describe(driver, sub)))
+            .FullReference.ShouldBe("Tank01.Level");
+
+        sub.DroppedInitialValues.ShouldBe(0);
+    }
+
+    /// <summary>A first ReadAsync on a fresh driver (no pump yet) requests the stream before its advise.</summary>
+    [Fact]
+    public async Task First_read_advises_only_after_the_stream_is_requested()
+    {
+        var sub = new GatewayLikeSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub)
+        {
+            ReadViaSubscribeDeadline = TimeSpan.FromSeconds(10),
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        driver.HasEventPumpForTests.ShouldBeFalse("precondition: the read is the first advise");
+
+        var read = await driver.ReadAsync(["Tank02.Level"], CancellationToken.None);
+
+        var calls = sub.CallsSnapshot();
+        Array.IndexOf(calls, "stream").ShouldBe(0, $"the stream must be requested before the read's advise; calls=[{string.Join(" | ", calls)}]");
+        read[0].StatusCode.ShouldBe(0u, "the read's initial value must arrive (not BadTimeout)");
+        sub.DroppedInitialValues.ShouldBe(0);
+    }
+
+    /// <summary>
+    ///     The replay's re-advise of DATA subscriptions (which has no per-advise hook) waits for the restarted
+    ///     pump's stream, so each re-advised tag's new initial value arrives. The value flips so only a
+    ///     post-replay event can satisfy the wait.
+    /// </summary>
+    [Fact]
+    public async Task Reconnect_replay_delivers_re_advised_data_subscriptions_initial_values()
+    {
+        var sub = new GatewayLikeSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub);
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        var last = new System.Collections.Concurrent.ConcurrentDictionary<string, object?>();
+        driver.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot.Value;
+
+        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
+        await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is true,
+            "positive control: the subscribe's initial value arrives", () => Describe(driver, sub));
+
+        sub.ScanStateRunning = false; // the fake's next initial value
+        await driver.InvokeReplayForTestAsync(CancellationToken.None);
+
+        await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is false,
+            "the replay's re-advise must deliver its new initial value",
+            () => Describe(driver, sub) + $" last=[{string.Join(",", last.Select(kv => $"{kv.Key}={kv.Value}"))}] parked={driver.ParkedEarlyEventCountForTests} h2=[{string.Join(",", driver.SubscriptionsForTests.ResolveSubscribers(2).Select(b => b.FullReference))}] h1=[{string.Join(",", driver.SubscriptionsForTests.ResolveSubscribers(1).Select(b => b.FullReference))}]");
+        sub.DroppedInitialValues.ShouldBe(0);
+    }
+
+    /// <summary>
+    ///     A replay whose restarted pump's stream ends at once must FAIL (the supervisor then backs off and
+    ///     stays Degraded) rather than succeed onto a dead stream — succeeding went Healthy, and the next
+    ///     advise's re-report started a new cycle with no backoff: a tight reconnect loop. Positive control:
+    ///     the first replay, whose stream stays open, succeeds.
+    /// </summary>
+    [Fact]
+    public async Task Replay_fails_when_the_restarted_stream_ends_at_once()
+    {
+        var sub = new GatewayLikeSubscriber { StreamsThatStayOpen = 2 };
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub);
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); // stream 1
+
+        await driver.InvokeReplayForTestAsync(CancellationToken.None); // stream 2 stays open — succeeds
+
+        await Should.ThrowAsync<InvalidOperationException>(
+            () => driver.InvokeReplayForTestAsync(CancellationToken.None), // stream 3 ends at once
+            "a replay must not report success over a dead event stream");
+    }
+
+    /// <summary>
+    ///     #865 review: a stream fault is escalated only from the LIVE pump. A fault from a pump superseded
+    ///     by a reopen (its session is no longer current) is ignored — it would fail a good replay — while
+    ///     the live pump's faults (before the reopen, and the replay's new pump after it) still escalate.
+    /// </summary>
+    [Fact]
+    public async Task Only_the_live_pumps_stream_faults_are_escalated()
+    {
+        var sub = new GatewayLikeSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub)
+        {
+            RecreateOverrideForTests = _ => Task.CompletedTask,
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); // pump 1
+
+        // Positive control: the live pump's fault escalates.
+        sub.FaultAttachedStream(new IOException("drop 1"));
+        await TestWait.UntilAsync(() => driver.EscalatedStreamFaultsForTests == 1, "the live pump's fault must escalate");
+
+        // Recover onto pump 2 (the replay restarts the faulted pump), then reopen: pump 2 is now superseded.
+        await driver.InvokeReplayForTestAsync(CancellationToken.None);
+        var pump2 = driver.EventPumpInstanceForTests.ShouldNotBeNull();
+        await driver.InvokeReopenForTestAsync(CancellationToken.None);
+        sub.FaultAttachedStream(new IOException("late fault from the superseded pump"));
+        await TestWait.UntilAsync(() => ((EventPump)pump2).HasEnded, "the superseded pump's stream must have faulted");
+
+        // Positive control after the absence: the replay's new pump faults and that DOES escalate — so the
+        // count can only be 2 if the superseded pump's fault was ignored.
+        await driver.InvokeReplayForTestAsync(CancellationToken.None);
+        var pump3 = (EventPump)driver.EventPumpInstanceForTests.ShouldNotBeNull();
+        pump3.ShouldNotBeSameAs(pump2);
+        sub.FaultAttachedStream(new IOException("drop 3"));
+        // HasEnded is set only after the fault callback ran, so once both pumps have ended every escalation
+        // that will ever happen has happened.
+        await TestWait.UntilAsync(() => pump3.HasEnded, "the new live pump's stream must have faulted");
+        driver.EscalatedStreamFaultsForTests.ShouldBe(2, "the new live pump's fault escalates; the superseded pump's did not");
+    }
+
+    /// <summary>
+    ///     A probe sync on a driver whose pump was never started (no platform source, e.g. the
+    ///     DiscoverAsync path) still starts it before advising — the watcher's pre-subscribe hook, not just
+    ///     the init path.
+    /// </summary>
+    [Fact]
+    public async Task Probe_sync_without_a_platform_source_still_starts_the_stream_first()
+    {
+        var sub = new GatewayLikeSubscriber();
+        // No platform source: init starts no pump; the probe set arrives through SyncPlatformsAsync.
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub);
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        driver.HasEventPumpForTests.ShouldBeFalse("precondition: nothing has started the pump yet");
+
+        await driver.ProbeWatcherForTests.ShouldNotBeNull().SyncPlatformsAsync(Platforms, CancellationToken.None);
+
+        await TestWait.UntilAsync(
+            () => Platforms.All(p => StateOf(driver, p) == HostState.Running),
+            "the watcher must start the stream before advising, so the initial values arrive",
+            () => Describe(driver, sub));
+        sub.DroppedInitialValues.ShouldBe(0);
+    }
+}
@@ -15,6 +15,85 @@ public sealed class ReconnectSupervisorTests
        new(InitialBackoffOverride: TimeSpan.FromMilliseconds(5),
            MaxBackoffOverride: TimeSpan.FromMilliseconds(20));

+    /// <summary>
+    ///     #865: AttemptInProgress is true while a reopen (through the replay) runs and FALSE during the backoff
+    ///     between attempts — when the state still reads Reopening. Advises wait for a replay only while one is
+    ///     actually running, not for a whole backoff.
+    /// </summary>
+    [Fact]
+    public async Task AttemptInProgress_is_true_during_an_attempt_and_false_during_the_backoff()
+    {
+        var reopenEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseReopen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var reopens = 0;
+        using var sup = new ReconnectSupervisor(
+            reopen: async _ =>
+            {
+                if (Interlocked.Increment(ref reopens) == 1)
+                {
+                    reopenEntered.TrySetResult();
+                    await releaseReopen.Task;
+                    throw new IOException("gateway still down");
+                }
+            },
+            replay: _ => Task.CompletedTask,
+            backoffDelay: (_, _, _) => TimeSpan.FromMinutes(10)); // park in the backoff
+
+        sup.AttemptInProgress.ShouldBeFalse("Healthy: no attempt");
+        sup.ReportTransportFailure(new IOException("drop"));
+        await TestWait.ForAsync(reopenEntered.Task, "the first reopen must start");
+        sup.AttemptInProgress.ShouldBeTrue("a reopen is running");
+
+        releaseReopen.SetResult(); // the reopen fails -> backoff of 10 min
+        await TestWait.UntilAsync(() => !sup.AttemptInProgress, "the failed attempt must clear the flag");
+        sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Reopening, "the backoff still reads Reopening — why the flag exists");
+    }
+
+    /// <summary>
+    ///     #865 review: a transport failure reported WHILE the replay runs (e.g. the restarted event stream is
+    ///     refused a round trip later) must fail that attempt — not go Healthy over a dead transport. The
+    ///     supervisor retries (with backoff) and goes Healthy only after a clean replay. Without the fix the
+    ///     first replay's success went Healthy and replay ran once.
+    /// </summary>
+    [Fact]
+    public async Task A_failure_reported_during_the_replay_fails_that_attempt()
+    {
+        ReconnectSupervisor? self = null;
+        var replays = 0;
+        using var sup = new ReconnectSupervisor(
+            reopen: _ => Task.CompletedTask,
+            replay: _ =>
+            {
+                if (Interlocked.Increment(ref replays) == 1)
+                {
+                    self!.ReportTransportFailure(new IOException("stream refused during replay"));
+                }
+                return Task.CompletedTask;
+            },
+            options: FastBackoff());
+        self = sup;
+
+        sup.ReportTransportFailure(new IOException("drop"));
+        await TestWait.HealthyAsync(sup, "the second, clean replay must recover");
+        Volatile.Read(ref replays).ShouldBe(2, "the faulted replay must not count as recovered");
+    }
+
+    /// <summary>#865: a transport failure reported after StopAsync never starts a recovery loop (it throws ODE).</summary>
+    [Fact]
+    public async Task ReportTransportFailure_after_stop_never_starts_a_loop()
+    {
+        var reopens = 0;
+        var sup = new ReconnectSupervisor(
+            reopen: _ => { Interlocked.Increment(ref reopens); return Task.CompletedTask; },
+            replay: _ => Task.CompletedTask,
+            options: FastBackoff());
+        await sup.StopAsync();
+
+        Should.Throw<ObjectDisposedException>(() => sup.ReportTransportFailure(new IOException("late")));
+        sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Healthy);
+        Volatile.Read(ref reopens).ShouldBe(0);
+    }
+
    /// <summary>Verifies that the supervisor starts in a healthy state with no errors.</summary>
    [Fact]
    public void InitialState_IsHealthy()