feat(galaxy): wait on the gateway's stream-attached ack instead of the 1 s best-effort wait (#877)

mxaccessgw#227 (Contracts/Client 0.4.0) adds an opt-in attach acknowledgement on
StreamEvents: a StreamAttached sentinel MxEvent sent first, only after the
stream's subscriber is registered on the session. This closes the #865 attach
race that waiting for "stream requested" only narrowed.

- Bump vendored MxGateway.Client/Contracts 0.3.0 -> 0.4.0 (nuget-packages/,
  pulled from the Gitea feed; local-mxgw stays the mapped source).
- IGalaxySubscriber.SupportsStreamAttachedAck (default false, so fakes and older
  seams keep the fallback); GatewayGalaxySubscriber reads it from the live
  session's capabilities (stream-events-attached-ack) and passes
  emitStreamAttached accordingly; TracedGalaxySubscriber forwards it.
- EventPump: records AttachAckExpected before the stream opens (published
  before StreamRequested completes); intercepts ev.StreamAttached ahead of the
  ReplayGap check - completes StreamAttached and continues: never dispatched,
  never counted as received, never a resume cursor (worker_sequence 0). The
  signal also completes when the loop ends so no waiter hangs. One pump = one
  stream, so a reconnect's new pump has a fresh signal. New counter
  galaxy.events.stream_attached.
- AwaitCurrentEventStreamAsync: with the capability, waits on StreamAttached
  within the same StreamRequestWait budget (1 s); a missing ack logs a
  rate-limited Warning and proceeds instead of hanging. Without the capability
  the old StreamRequested wait is the fallback.

Tests (StreamAttachedAckTests, 6): ack-gated subscribe gets its initial value;
capability absent uses the fallback; missing ack times out with a warning; the
sentinel is never dispatched; a dead stream releases ack waiters; reconnect
gives each stream a fresh signal. With the ack wait skipped, 3 of them fail.

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

Directory.Packages.props

@@ -230,8 +230,8 @@
     <PackageVersion Include="ZB.MOM.WW.Health.EntityFrameworkCore" Version="0.3.0" />
     <PackageVersion Include="ZB.MOM.WW.Telemetry" Version="0.1.0" />
     <PackageVersion Include="ZB.MOM.WW.Telemetry.Serilog" Version="0.1.0" />
-    <PackageVersion Include="ZB.MOM.WW.MxGateway.Client" Version="0.3.0" />
-    <PackageVersion Include="ZB.MOM.WW.MxGateway.Contracts" Version="0.3.0" />
+    <PackageVersion Include="ZB.MOM.WW.MxGateway.Client" Version="0.4.0" />
+    <PackageVersion Include="ZB.MOM.WW.MxGateway.Contracts" Version="0.4.0" />
     <PackageVersion Include="ZB.MOM.WW.Configuration" Version="0.1.0" />
     <PackageVersion Include="ZB.MOM.WW.Auth.Abstractions" Version="0.4.1" />
     <PackageVersion Include="ZB.MOM.WW.Auth.Ldap" Version="0.4.1" />

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

@@ -2213,9 +2213,11 @@ public sealed class GalaxyDriver
     }
 
     /// <summary>
-    ///     Upper bound on how long an advise waits for a current-session event stream (#865). Kept well under
-    ///     Galaxy's Tier-A Read (2 s) / Subscribe (5 s) call budgets, so the "proceed anyway" fallback is
-    ///     reachable before the resilience pipeline times the call out.
+    ///     Upper bound on how long an advise waits for a current-session event stream (#865) — including, on a
+    ///     gateway with the stream-attached ack, the ack itself (#877). Kept well under Galaxy's Tier-A Read (2 s) /
+    ///     Subscribe (5 s) call budgets, so the "proceed anyway" fallback is reachable before the resilience
+    ///     pipeline times the call out. The ack normally lands within one round trip, so a timeout on that path is
+    ///     a fault worth a warning, not a routine fallback.
     /// </summary>
     internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(1);
 
@@ -2227,9 +2229,14 @@ public sealed class GalaxyDriver
     ///     <para>
     ///     A fresh StreamEvents (after_worker_sequence = 0) is never replayed, so an initial value pushed before
     ///     the stream attaches is lost — and a slow-changing attribute (ScanState, a setpoint) may not publish
-    ///     again for a long time. So: start the pump if there is none, and wait until its stream call has started
-    ///     (<see cref="EventPump.StreamRequested"/> — this narrows, it cannot close, the attach race: the gateway
-    ///     offers no attach acknowledgement).
+    ///     again for a long time. So: start the pump if there is none, and wait until its stream is attached.
+    ///     </para>
+    ///     <para>
+    ///     #877 — when the gateway advertised <c>stream-events-attached-ack</c> (<see cref="EventPump.AttachAckExpected"/>),
+    ///     "attached" is deterministic: the wait is on <see cref="EventPump.StreamAttached"/>, the gateway's sentinel
+    ///     that the stream is registered on the session, so the advise cannot lose its initial value. An older
+    ///     gateway has no ack; the wait falls back to <see cref="EventPump.StreamRequested"/> (the stream call has
+    ///     started), which only narrows the attach race.
     ///     </para>
     ///     <para>
     ///     A pump bound to an OLDER session belongs to a gateway session a reconnect replaced (the reopen →
@@ -2284,6 +2291,22 @@ public sealed class GalaxyDriver
                     try
                     {
                         await pump.StreamRequested.WaitAsync(remaining, cancellationToken).ConfigureAwait(false);
+                        // AttachAckExpected is published before StreamRequested completes (#877).
+                        if (pump.AttachAckExpected && !pump.HasEnded)
+                        {
+                            remaining = StreamRequestWait - elapsed.Elapsed;
+                            if (remaining < TimeSpan.Zero) remaining = TimeSpan.Zero;
+                            try
+                            {
+                                await pump.StreamAttached.WaitAsync(remaining, cancellationToken).ConfigureAwait(false);
+                            }
+                            catch (TimeoutException)
+                            {
+                                if (pump.HasEnded) ReportEndedPump(pump);
+                                LogNoStreamAttachedAck();
+                                return session;
+                            }
+                        }
                         if (pump.HasEnded) ReportEndedPump(pump);
                         return session;
                     }
@@ -2292,8 +2315,10 @@ public sealed class GalaxyDriver
                         // fall through to the log
                     }
                 }
-                else if (pump.StreamRequested.IsCompleted)
+                else if (pump.StreamRequested.IsCompleted
+                         && (!pump.AttachAckExpected || pump.StreamAttached.IsCompleted))
                 {
+                    if (pump.HasEnded) ReportEndedPump(pump);
                     return session;
                 }
             }
@@ -2340,6 +2365,30 @@ public sealed class GalaxyDriver
     }
 
     private long _lastNoStreamWarningTicks;
+    private long _lastNoAckWarningTicks;
+
+    /// <summary>
+    ///     #877 — the gateway advertised the stream-attached ack but did not send it within
+    ///     <see cref="StreamRequestWait"/>. Loud (Warning, once a minute) because the capability promises the
+    ///     sentinel as the stream's first message: its absence means a slow/stuck attach or a gateway bug, and the
+    ///     advise proceeding now may lose its initial value.
+    /// </summary>
+    private void LogNoStreamAttachedAck()
+    {
+        var now = Environment.TickCount64;
+        var last = Interlocked.Read(ref _lastNoAckWarningTicks);
+        if ((last == 0 || now - last >= 60_000)
+            && Interlocked.CompareExchange(ref _lastNoAckWarningTicks, now, last) == last)
+        {
+            _logger.LogWarning(
+                "GalaxyDriver {InstanceId}: the gateway advertises stream-events-attached-ack but its event stream sent no attach acknowledgement within {Wait} — advising anyway; initial values may be missed.",
+                _driverInstanceId, StreamRequestWait);
+            return;
+        }
+        _logger.LogDebug(
+            "GalaxyDriver {InstanceId} advising without the event stream's attach acknowledgement (warning rate-limited).",
+            _driverInstanceId);
+    }
 
     /// <summary>Warning at most once a minute while the supervisor is Healthy; Debug otherwise (an outage would flood).</summary>
     private void LogNoCurrentStream()

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

@@ -51,6 +51,9 @@ internal sealed class EventPump : IAsyncDisposable
     private static readonly Counter<long> ReplayGapsObserved =
         Meter.CreateCounter<long>("galaxy.events.replay_gaps", unit: "{gap}",
             description: "Replay-gap sentinels received — the gateway evicted events we can never see.");
+    private static readonly Counter<long> StreamAttachedAcks =
+        Meter.CreateCounter<long>("galaxy.events.stream_attached", unit: "{ack}",
+            description: "Stream-attached sentinels received — the gateway confirmed the event stream is registered (#877).");
     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.");
@@ -175,10 +178,38 @@ internal sealed class EventPump : IAsyncDisposable
     ///     ⚠️ This NARROWS the stream-attach race, it does not close it: the HTTP/2 request is sent
     ///     asynchronously and the gateway attaches the subscriber after auth, concurrently with any other
     ///     RPC. In practice the attach wins (an advise's initial value needs a worker IPC round trip plus an
-    ///     MXAccess advise), but the gateway offers no attach acknowledgement to await.
+    ///     MXAccess advise). A gateway with the <c>stream-events-attached-ack</c> capability (mxaccessgw#227)
+    ///     closes it: wait on <see cref="StreamAttached"/> when <see cref="AttachAckExpected"/> (#877).
     /// </remarks>
     public Task StreamRequested => _streamRequested.Task;
 
+    private readonly TaskCompletionSource _streamAttached = new(TaskCreationOptions.RunContinuationsAsynchronously);
+    private volatile bool _attachAckExpected;
+
+    /// <summary>
+    ///     #877 — true when this pump's stream was requested WITH the gateway's stream-attached acknowledgement
+    ///     (the subscriber reported <see cref="IGalaxySubscriber.SupportsStreamAttachedAck"/> when the stream was
+    ///     opened). Set before <see cref="StreamRequested"/> completes, so a caller that awaited that first reads
+    ///     the final value.
+    /// </summary>
+    public bool AttachAckExpected => _attachAckExpected;
+
+    /// <summary>
+    ///     #877 — completes when the gateway's stream-attached sentinel arrives: the stream is registered on the
+    ///     gateway session, so every event the session publishes from then on (an advise's initial value
+    ///     included) reaches this stream. This CLOSES the attach race that <see cref="StreamRequested"/> only
+    ///     narrows. Also completes once the loop has ended, so a waiter never hangs on a dead pump — check
+    ///     <see cref="HasEnded"/> afterwards. Never completes for a stream opened without the ack
+    ///     (<see cref="AttachAckExpected"/> false) while it is alive: callers wait on it only when expected.
+    ///     One pump owns exactly one stream; a reconnect replaces the pump, so every new stream starts with a
+    ///     fresh, uncompleted signal.
+    /// </summary>
+    public Task StreamAttached => _streamAttached.Task;
+
+    /// <summary>True once the stream-attached sentinel itself arrived (not merely the loop ending).</summary>
+    internal bool StreamAttachedReceived => Volatile.Read(ref _attachAckReceived);
+    private bool _attachAckReceived;
+
     /// <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.
@@ -213,6 +244,9 @@ internal sealed class EventPump : IAsyncDisposable
         IAsyncEnumerator<MxEvent>? enumerator = null;
         try
         {
+            // #877: read the capability BEFORE opening the stream — the subscriber requests the ack from the
+            // same session state — and publish it before StreamRequested releases a waiter.
+            _attachAckExpected = _subscriber.SupportsStreamAttachedAck;
             enumerator = _subscriber.StreamEventsAsync(ct).GetAsyncEnumerator(ct);
             // Start the first MoveNext BEFORE awaiting it: that synchronous step issues the RPC, which is
             // the moment StreamRequested reports (#865).
@@ -234,6 +268,17 @@ internal sealed class EventPump : IAsyncDisposable
             {
                 var ev = enumerator.Current;
                 if (ct.IsCancellationRequested) break;
+
+                // #877: the stream-attached sentinel is a control signal (family UNSPECIFIED, no body,
+                // worker_sequence 0). It is handled before anything else: it is never dispatched, never
+                // counted as a received/unknown event, and never touches a resume cursor (this driver
+                // keeps none — it always opens fresh streams — and a 0 sequence must never become one).
+                if (ev.StreamAttached is not null)
+                {
+                    HandleStreamAttached(ev.StreamAttached);
+                    continue;
+                }
+
                 EventsReceived.Add(1, _clientTag);
 
                 // The replay-gap sentinel is a control signal, not data, and is handled HERE
@@ -294,6 +339,8 @@ internal sealed class EventPump : IAsyncDisposable
             Volatile.Write(ref _ended, true);
             // A loop that died before issuing the RPC must not strand a StreamRequested waiter.
             _streamRequested.TrySetResult();
+            // Nor one waiting for an attach acknowledgement that can no longer arrive (#877).
+            _streamAttached.TrySetResult();
             if (enumerator is not null)
             {
                 try { await enumerator.DisposeAsync().ConfigureAwait(false); }
@@ -358,6 +405,25 @@ internal sealed class EventPump : IAsyncDisposable
     ///     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.
     /// </remarks>
+    private void HandleStreamAttached(ZB.MOM.WW.MxGateway.Contracts.Proto.StreamAttached attached)
+    {
+        StreamAttachedAcks.Add(1, _clientTag);
+        if (!_attachAckExpected)
+        {
+            // An unrequested ack (capability read raced a session swap, or a gateway that sends it anyway)
+            // is harmless — still a control signal, still never dispatched.
+            _logger.LogDebug("Galaxy event stream sent a stream-attached acknowledgement that was not expected — ignoring it.");
+        }
+        else
+        {
+            _logger.LogDebug(
+                "Galaxy event stream attached on the gateway (live after worker_sequence {LiveAfter}).",
+                attached.LiveAfterSequence);
+        }
+        Volatile.Write(ref _attachAckReceived, true);
+        _streamAttached.TrySetResult();
+    }
+
     private void HandleReplayGap(ReplayGap gap)
     {
         ReplayGapsObserved.Add(1, _clientTag);

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

@@ -158,6 +158,15 @@ public sealed class GatewayGalaxySubscriber : IGalaxySubscriber
         var session = _session.Session
             ?? throw new InvalidOperationException(
                 "GalaxyMxSession is not connected. StreamEventsAsync called before ConnectAsync.");
-        return session.StreamEventsAsync(afterWorkerSequence: 0, cancellationToken);
+        // #877: ask for the stream-attached sentinel only when the gateway advertised it — an older
+        // gateway ignores the field, so the pump would wait for an ack that never comes.
+        return session.StreamEventsAsync(
+            afterWorkerSequence: 0,
+            emitStreamAttached: session.SupportsStreamAttachedAck,
+            cancellationToken);
     }
+
+    /// <inheritdoc />
+    /// <remarks>Read from the live session's <c>OpenSessionReply.capabilities</c>; false while disconnected.</remarks>
+    public bool SupportsStreamAttachedAck => _session.Session?.SupportsStreamAttachedAck ?? false;
 }

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

@@ -38,4 +38,15 @@ public interface IGalaxySubscriber
     /// <param name="cancellationToken">Cancellation token for the stream.</param>
     /// <returns>An asynchronous stream of <see cref="MxEvent"/>s as they arrive from the gateway.</returns>
     IAsyncEnumerable<MxEvent> StreamEventsAsync(CancellationToken cancellationToken);
+
+    /// <summary>
+    ///     #877 — true when the stream returned by <see cref="StreamEventsAsync"/> opens with the gateway's
+    ///     stream-attached sentinel (<see cref="MxEvent.StreamAttached"/>, mxaccessgw#227): the session
+    ///     advertised the <c>stream-events-attached-ack</c> capability and the subscriber requests the ack.
+    ///     The sentinel means the stream is registered on the gateway session, so an advise issued after it
+    ///     cannot lose its initial value to the attach race (#865). False (the default) for an older gateway
+    ///     and for test fakes that do not model the ack — the <see cref="EventPump"/> then falls back to the
+    ///     best-effort "stream requested" wait.
+    /// </summary>
+    bool SupportsStreamAttachedAck => false;
 }

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

@@ -49,6 +49,9 @@ internal sealed class TracedGalaxySubscriber(IGalaxySubscriber inner, string cli
         }
     }
 
+    /// <inheritdoc />
+    public bool SupportsStreamAttachedAck => inner.SupportsStreamAttachedAck;
+
     // Streaming RPC — one parent span covers the entire stream lifetime. Per-event
     // spans would dominate the trace volume at 50k tags / 1Hz; ops gets per-event
     // visibility through EventPump's metrics instead.

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

@@ -0,0 +1,381 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using Google.Protobuf.WellKnownTypes;
+using Microsoft.Extensions.Logging;
+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.Config;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
+
+/// <summary>
+///     #877 — the gateway's stream-attached acknowledgement (mxaccessgw#227). With the
+///     <c>stream-events-attached-ack</c> capability, an advise waits until the gateway confirms the event
+///     stream is REGISTERED on the session, closing the #865 attach race that the "stream requested" wait only
+///     narrowed. Without it the old best-effort wait stays the fallback.
+/// </summary>
+/// <remarks>
+///     The fake models the race the real gateway has: a stream call is issued at once but the gateway
+///     registers its subscriber only later (here: when the test opens the per-stream attach gate). An advise's
+///     initial value pushed before that registration is lost, exactly as on the gateway — so an advise that
+///     does not wait for the ack loses its value, and one that does cannot.
+/// </remarks>
+public sealed class StreamAttachedAckTests
+{
+    private static GalaxyDriverOptions Opts() => new(
+        new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
+        new GalaxyMxAccessOptions("StreamAttachedAck"),
+        new GalaxyRepositoryOptions(WatchDeployEvents: false),
+        new GalaxyReconnectOptions());
+
+    private static MxEvent AckSentinel(ulong liveAfter = 0) => new()
+    {
+        // Shaped as the gateway emits it: no family, no body, worker_sequence 0.
+        Family = MxEventFamily.Unspecified,
+        StreamAttached = new StreamAttached { LiveAfterSequence = liveAfter },
+    };
+
+    /// <summary>
+    ///     Gateway-like subscriber whose stream registration is DELAYED until the test opens that stream's
+    ///     attach gate. Records "stream:N", "attached:N" and "subscribe:&lt;refs&gt;" in order.
+    /// </summary>
+    private sealed class DelayedAttachSubscriber(bool supportsAck, bool sendsAck = true) : IGalaxySubscriber
+    {
+        private readonly Lock _gate = new();
+        private readonly List<TaskCompletionSource> _attachGates = [];
+        private Channel<MxEvent>? _attached;
+        private int _nextHandle;
+        private int _streams;
+
+        public bool SupportsStreamAttachedAck => supportsAck;
+
+        public int DroppedInitialValues;
+        public volatile bool NextValue = true;
+
+        public List<string> Calls { get; } = [];
+
+        public string[] CallsSnapshot() { lock (_gate) return [.. Calls]; }
+
+        /// <summary>Lets stream N (1-based) register on the "gateway".</summary>
+        public void Attach(int stream) => GateFor(stream).TrySetResult();
+
+        private TaskCompletionSource GateFor(int stream)
+        {
+            lock (_gate)
+            {
+                while (_attachGates.Count < stream)
+                    _attachGates.Add(new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
+                return _attachGates[stream - 1];
+            }
+        }
+
+        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,
+                        WorkerSequence = (ulong)h,
+                        Value = new MxValue { BoolValue = NextValue },
+                        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)
+        {
+            var n = Interlocked.Increment(ref _streams);
+            lock (_gate) Calls.Add("stream:" + n);
+            // The gateway registers the subscriber only after auth etc. — asynchronously to the call.
+            await GateFor(n).Task.WaitAsync(cancellationToken).ConfigureAwait(false);
+            var mine = Channel.CreateUnbounded<MxEvent>();
+            lock (_gate)
+            {
+                _attached = mine;
+                Calls.Add("attached:" + n);
+            }
+            try
+            {
+                // The gateway sends the sentinel only AFTER registration, and only when asked.
+                if (supportsAck && sendsAck) yield return AckSentinel();
+                await foreach (var ev in mine.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+                {
+                    yield return ev;
+                }
+            }
+            finally
+            {
+                lock (_gate) { if (ReferenceEquals(_attached, mine)) _attached = null; }
+            }
+        }
+    }
+
+    private static string Describe(DelayedAttachSubscriber sub) =>
+        $"calls=[{string.Join(" | ", sub.CallsSnapshot())}] dropped={sub.DroppedInitialValues}";
+
+    /// <summary>
+    ///     The key test. Capability present: the subscribe is NOT issued while the stream is requested but not
+    ///     yet registered; it goes out only after the ack, so its initial value arrives. Fail-before: with the
+    ///     ack wait skipped (pre-#877 behaviour) the subscribe goes out at "stream requested", its initial value
+    ///     is pushed to no stream, and this test fails on the ordering and the dropped count.
+    /// </summary>
+    [Fact]
+    public async Task Capability_present_subscribe_waits_for_the_ack_and_gets_its_initial_value()
+    {
+        var sub = new DelayedAttachSubscriber(supportsAck: true);
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub)
+        {
+            StreamRequestWait = TimeSpan.FromSeconds(20), // generous: the gate below decides, not the budget
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        var delivered = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
+        driver.OnDataChange += (_, e) => delivered.TrySetResult(e);
+
+        var subscribe = driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
+
+        await TestWait.UntilAsync(() => sub.CallsSnapshot().Contains("stream:1"),
+            "positive control: the subscribe must start the event stream", () => Describe(sub));
+        // The stream is requested but not registered. Give a skipped wait every chance to advise now.
+        await Task.Delay(300, TestContext.Current.CancellationToken);
+        sub.CallsSnapshot().ShouldNotContain(c => c.StartsWith("subscribe:", StringComparison.Ordinal),
+            $"no advise may be issued before the gateway acknowledged the stream; {Describe(sub)}");
+        subscribe.IsCompleted.ShouldBeFalse("the subscribe must still be waiting for the attach ack");
+
+        sub.Attach(1);
+        await TestWait.ForAsync(subscribe, "the subscribe proceeds once the ack arrives", () => Describe(sub));
+
+        var calls = sub.CallsSnapshot();
+        Array.IndexOf(calls, "attached:1").ShouldBeLessThan(
+            Array.FindIndex(calls, c => c.StartsWith("subscribe:", StringComparison.Ordinal)),
+            $"the advise must follow the registration; {Describe(sub)}");
+        (await TestWait.ForAsync(delivered.Task, "the initial value must arrive", () => Describe(sub)))
+            .FullReference.ShouldBe("Tank01.Level");
+        sub.DroppedInitialValues.ShouldBe(0);
+        ((EventPump)driver.EventPumpInstanceForTests!).StreamAttachedReceived.ShouldBeTrue();
+    }
+
+    /// <summary>
+    ///     Capability absent (an older gateway): no ack is requested or awaited — the advise proceeds as soon as
+    ///     the stream call has started (the #865 best-effort wait), even though the stream is not registered yet.
+    ///     That leaves the narrowed race in place (the value is dropped here, by construction of the fake); the
+    ///     point is that the driver does not hang waiting for an ack an old gateway never sends.
+    /// </summary>
+    [Fact]
+    public async Task Capability_absent_falls_back_to_the_stream_requested_wait()
+    {
+        var sub = new DelayedAttachSubscriber(supportsAck: false);
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub)
+        {
+            StreamRequestWait = TimeSpan.FromSeconds(20),
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+
+        var sw = System.Diagnostics.Stopwatch.StartNew();
+        await TestWait.ForAsync(
+            driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None),
+            "without the capability the subscribe must not wait for an ack", () => Describe(sub));
+        sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), "the 20 s budget must not have been spent on an ack");
+
+        var pump = (EventPump)driver.EventPumpInstanceForTests!;
+        pump.AttachAckExpected.ShouldBeFalse();
+        pump.StreamAttached.IsCompleted.ShouldBeFalse("no ack was requested, none arrived");
+        var calls = sub.CallsSnapshot();
+        Array.IndexOf(calls, "stream:1").ShouldBe(0, $"the stream is still requested first; {Describe(sub)}");
+        calls.ShouldContain(c => c.StartsWith("subscribe:", StringComparison.Ordinal));
+        calls.ShouldNotContain("attached:1", "the subscribe went out before registration — the old, narrowed race");
+        sub.Attach(1); // let the stream finish attaching so disposal is clean
+    }
+
+    /// <summary>
+    ///     Capability advertised but the ack never comes (a gateway bug or a stuck attach): the advise does not
+    ///     hang — it proceeds after <see cref="GalaxyDriver.StreamRequestWait"/> and logs a Warning naming the
+    ///     missing acknowledgement.
+    /// </summary>
+    [Fact]
+    public async Task Missing_ack_times_out_with_a_warning_instead_of_hanging()
+    {
+        var sub = new DelayedAttachSubscriber(supportsAck: true, sendsAck: false);
+        var log = new CapturingLogger();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub,
+            logger: new TypedLogger(log))
+        {
+            StreamRequestWait = TimeSpan.FromMilliseconds(300),
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        sub.Attach(1); // registered, but no sentinel is sent
+
+        await TestWait.ForAsync(
+            driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None),
+            "a missing ack must not hang the subscribe", () => Describe(sub));
+
+        lock (log.Entries)
+        {
+            log.Entries.ShouldContain(e => e.Level == LogLevel.Warning && e.Text.Contains("attach acknowledgement", StringComparison.Ordinal),
+                $"a missing ack must be logged loudly; log=\n{log.AllText}");
+        }
+        sub.CallsSnapshot().ShouldContain(c => c.StartsWith("subscribe:", StringComparison.Ordinal));
+    }
+
+    /// <summary>
+    ///     The sentinel is a control signal: never dispatched, never counted as a received event, and it
+    ///     completes <see cref="EventPump.StreamAttached"/>. Positive control: a real event after it IS dispatched.
+    /// </summary>
+    [Fact]
+    public async Task Ack_sentinel_is_never_dispatched()
+    {
+        var subscriber = new ChannelSubscriber();
+        var registry = new SubscriptionRegistry();
+        registry.Register(1, [new TagBinding("Tank.Level", ItemHandle: 7)]);
+
+        var dispatched = 0;
+        var realValue = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
+        await using var pump = new EventPump(subscriber, registry, channelCapacity: 8, clientName: "AckTest");
+        pump.OnDataChange += (_, args) => { Interlocked.Increment(ref dispatched); realValue.TrySetResult(args); };
+        pump.Start();
+
+        // An ack shaped with OnDataChange + item handle 7 would be dispatched if it reached the channel —
+        // so this proves interception, not just "unknown family is dropped".
+        var ack = AckSentinel(liveAfter: 0);
+        ack.Family = MxEventFamily.OnDataChange;
+        ack.ItemHandle = 7;
+        ack.Value = new MxValue { DoubleValue = -1 };
+        await subscriber.EmitAsync(ack);
+        await TestWait.ForAsync(pump.StreamAttached, "the ack must complete the attached signal");
+        pump.StreamAttachedReceived.ShouldBeTrue();
+
+        await subscriber.EmitAsync(new MxEvent
+        {
+            Family = MxEventFamily.OnDataChange,
+            ItemHandle = 7,
+            Value = new MxValue { DoubleValue = 12.5 },
+            Quality = 192,
+            SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+        });
+
+        (await TestWait.ForAsync(realValue.Task, "positive control: real events still dispatch")).Snapshot.Value.ShouldBe(12.5);
+        Volatile.Read(ref dispatched).ShouldBe(1, "only the real event may be dispatched, never the ack");
+    }
+
+    /// <summary>
+    ///     A pump whose loop ends before any ack completes <see cref="EventPump.StreamAttached"/> (so no waiter
+    ///     hangs) WITHOUT reporting the ack as received.
+    /// </summary>
+    [Fact]
+    public async Task A_dead_stream_releases_ack_waiters_without_claiming_an_ack()
+    {
+        await using var pump = new EventPump(
+            new ThrowingSubscriber(), new SubscriptionRegistry(), onStreamFault: _ => { });
+        pump.Start();
+        await TestWait.ForAsync(pump.StreamAttached, "a dead pump must not strand an ack waiter");
+        pump.HasEnded.ShouldBeTrue();
+        pump.StreamAttachedReceived.ShouldBeFalse();
+        pump.AttachAckExpected.ShouldBeTrue();
+    }
+
+    /// <summary>
+    ///     Reconnect: the replay's new stream gets a FRESH attach signal — the first stream's ack does not
+    ///     release it. The replay's re-advise waits for the second stream's ack, and its new initial value
+    ///     arrives. Positive control: before stream 2 attaches, the re-advise has not been issued.
+    /// </summary>
+    [Fact]
+    public async Task Reconnect_gives_each_stream_a_fresh_signal()
+    {
+        var sub = new DelayedAttachSubscriber(supportsAck: true);
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub)
+        {
+            StreamRequestWait = TimeSpan.FromSeconds(20),
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        var last = new System.Collections.Concurrent.ConcurrentDictionary<string, object?>();
+        driver.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot.Value;
+
+        sub.Attach(1);
+        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 first stream delivers the initial value", () => Describe(sub));
+        var pump1 = (EventPump)driver.EventPumpInstanceForTests!;
+        pump1.StreamAttached.IsCompleted.ShouldBeTrue();
+
+        sub.NextValue = false;
+        var replay = driver.InvokeReplayForTestAsync(CancellationToken.None);
+        await TestWait.UntilAsync(() => sub.CallsSnapshot().Contains("stream:2"),
+            "the replay must start a new stream", () => Describe(sub));
+        var pump2 = (EventPump)driver.EventPumpInstanceForTests!;
+        pump2.ShouldNotBeSameAs(pump1);
+        pump2.StreamAttached.IsCompleted.ShouldBeFalse("a new stream starts with an uncompleted attach signal");
+
+        await Task.Delay(300, TestContext.Current.CancellationToken);
+        sub.CallsSnapshot().Count(c => c.StartsWith("subscribe:", StringComparison.Ordinal))
+            .ShouldBe(1, $"the re-advise must wait for stream 2's ack; {Describe(sub)}");
+
+        sub.Attach(2);
+        await TestWait.ForAsync(replay, "the replay completes once stream 2 is acknowledged", () => Describe(sub));
+        await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is false,
+            "the re-advised tag's new initial value must arrive on stream 2", () => Describe(sub));
+        sub.DroppedInitialValues.ShouldBe(0);
+    }
+
+    /// <summary>A stream fed directly by the test; reports the ack capability.</summary>
+    private sealed class ChannelSubscriber : IGalaxySubscriber
+    {
+        private readonly Channel<MxEvent> _stream = Channel.CreateUnbounded<MxEvent>();
+
+        public bool SupportsStreamAttachedAck => true;
+
+        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.Reader.ReadAllAsync(cancellationToken);
+
+        public ValueTask EmitAsync(MxEvent ev) => _stream.Writer.WriteAsync(ev);
+    }
+
+    private sealed class ThrowingSubscriber : IGalaxySubscriber
+    {
+        public bool SupportsStreamAttachedAck => true;
+
+        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)
+            => throw new InvalidOperationException("not connected");
+    }
+
+    private sealed class TypedLogger(CapturingLogger inner) : ILogger<GalaxyDriver>
+    {
+        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+            Func<TState, Exception?, string> formatter)
+        {
+            lock (inner.Entries) inner.Log(logLevel, eventId, state, exception, formatter);
+        }
+    }
+}