test(ab): replace fixed-delay absence assertions with positive controls / processed counts (#583)

Ab.Transport.Tests: predecessor-wait counter instead of 300 ms (DeviceSessionPoolTests x2), throttle queue/interval-park hooks instead of 50/100 ms (ForwardOpenThrottleTests x2), UnobservedTaskProbe with a must-report control fault instead of N GC cycles (EncapsulationClientTests), re-armed keep-alive timer as the control for the never-opened keep-alive test. AbCip.Tests: still-polling sibling / control driver read counts instead of 300-500 ms (subscriptions x4, alarm projection x5), structural ProbeTask==null + enabled-probe control (host probe x2), predecessor wait (NativeAbCipLeaseReleaseTests), processed waits in the shutdown/logging tests. AbLegacy.Tests: sibling-subscription control (Unsubscribe_halts_polling), ProbeCts==null + control. AbCip.IntegrationTests differential safety: a post-phase control open through the same throttle instead of 1.5-2.5 s sleeps (via a native tag init, keeping the harness read-only guard green). Fake tag ReadCount made interlocked; CapturingLogger entries snapshotted under a lock. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-07 23:28:33 -04:00
parent 3edff3304b
commit fe94a3bdb8
11 changed files with 224 additions and 116 deletions
@@ -399,6 +399,9 @@ public sealed class CipConnectionTests

        time.Advance(TimeSpan.FromSeconds(30));

+        // Positive control (#583): the keep-alive timer really ticked (at 8, 16 and 24 s) and re-armed itself, so
+        // the zero below is the timer declining to send, not a timer that never ran.
+        time.HasTimerDueAt(time.Start + TimeSpan.FromSeconds(32)).ShouldBeTrue();
        conn.KeepAlivesSent.ShouldBe(0);
        target.ForwardOpens.ShouldBe(0);
        target.Server.Connections.ShouldBe(0);
@@ -44,10 +44,10 @@ public sealed class DifferentialSafetyTests
    /// </summary>
    private static async Task ControlOpenAsync(DifferentialTarget target)
    {
-        var p = target.Params("@tags");
-        await using var lease = DeviceSessionPool.Shared.Acquire(NativeAbCipDevice.Key(p), NativeAbCipDevice.Options(p));
-        await Should.ThrowAsync<AbTransportException>(
-            () => lease.Scheduler.SendAsync([0x4C, 0x00], TestContext.Current.CancellationToken));
+        // A native tag initialisation (an existence read), like the harness's own: same pool, same device key.
+        using var runtime = AbCipTransportFactories.For(AbCipTransport.Native).Tags
+            .Create(target.Params("A", 1, false, AbCipDataType.DInt));
+        await Should.ThrowAsync<AbTransportException>(() => runtime.InitializeAsync(TestContext.Current.CancellationToken));
    }

    private static DeviceKey Key(DifferentialTarget target) => NativeAbCipDevice.Key(target.Params("@tags"));
@@ -48,26 +48,33 @@ public sealed class AbCipAlarmProjectionTests
        AbCipAlarmProjection.MapSeverity(900).ShouldBe(AlarmSeverity.Critical);
    }

+    private static AbCipDriverOptions Options(bool enabled) => new()
+    {
+        Devices = [new AbCipDeviceOptions(Device)],
+        RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
+        EnableAlarmProjection = enabled,
+        AlarmPollInterval = TimeSpan.FromMilliseconds(20),
+        // The ALMD projection here drives the parent-UDT runtime via offset-keyed values,
+        // so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003).
+        EnableDeclarationOnlyUdtGrouping = true,
+    };
+
    /// <summary>Verifies that disabled alarm projection returns a valid handle but does not poll.</summary>
    [Fact]
    public async Task FeatureFlag_Off_SubscribeAlarms_Returns_Handle_But_Never_Polls()
    {
        var factory = new FakeAbCipTagFactory();
-        var opts = new AbCipDriverOptions
-        {
-            Devices = [new AbCipDeviceOptions(Device)],
-            RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
-            EnableAlarmProjection = false, // explicit; also the default
-        };
-        var drv = new AbCipDriver(opts, "drv-1", factory);
+        var drv = new AbCipDriver(Options(enabled: false), "drv-1", factory);
        await drv.InitializeAsync("{}", CancellationToken.None);

        var handle = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
        handle.ShouldNotBeNull();
        handle.DiagnosticId.ShouldContain("abcip-alarm-sub-");

-        // Wait a touch — if polling were active, a fake member-read would be triggered.
-        await Task.Delay(100);
+        // Positive control (#583): the same options with the flag ON poll three times — so a loop this
+        // subscription had started would have read (and so created) the ALMD runtime by now.
+        await PolledThreeTimesWithTheFlagOnAsync();
+        factory.Tags.ShouldNotContainKey("HighTemp");
        factory.Tags.ShouldNotContainKey("HighTemp.InFaulted");
        factory.Tags.ShouldNotContainKey("HighTemp.Severity");

@@ -80,17 +87,7 @@ public sealed class AbCipAlarmProjectionTests
    public async Task FeatureFlag_On_Subscribe_Starts_Polling_And_Fires_Raise_On_0_to_1()
    {
        var factory = new FakeAbCipTagFactory();
-        var opts = new AbCipDriverOptions
-        {
-            Devices = [new AbCipDeviceOptions(Device)],
-            RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
-            EnableAlarmProjection = true,
-            AlarmPollInterval = TimeSpan.FromMilliseconds(20),
-            // The ALMD projection here drives the parent-UDT runtime via offset-keyed values,
-            // so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003).
-            EnableDeclarationOnlyUdtGrouping = true,
-        };
-        var drv = new AbCipDriver(opts, "drv-1", factory);
+        var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory);
        await drv.InitializeAsync("{}", CancellationToken.None);

        var events = new List<AlarmEventArgs>();
@@ -99,21 +96,16 @@ public sealed class AbCipAlarmProjectionTests
        var handle = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);

        // The ALMD UDT is declared so whole-UDT grouping kicks in; the parent HighTemp runtime
-        // gets created + polled. Set InFaulted offset-value to 0 first (clear), wait a tick,
-        // then flip to 1 (fault) + wait for the raise event.
+        // gets created + polled. Set InFaulted offset-value to 0 first (clear), let polls seed the
+        // "last-seen false" state, then flip to 1 (fault) + wait for the raise event.
        await WaitForTagCreation(factory, "HighTemp");
        factory.Tags["HighTemp"].ValuesByOffset[0] = 0;  // InFaulted=false at offset 0
        factory.Tags["HighTemp"].ValuesByOffset[8] = 500; // Severity at offset 8 (after InFaulted+Acked)
-        await Task.Delay(80); // let a tick seed the "last-seen false" state
+        await WaitForMoreReadsAsync(factory, "HighTemp", 2);

        factory.Tags["HighTemp"].ValuesByOffset[0] = 1; // flip to faulted
-        await Task.Delay(200); // allow several polls to be safe
-
-        lock (events)
-        {
-            events.ShouldContain(e => e.SourceNodeId == "HighTemp" && e.AlarmType == "ALMD"
-                && e.Message.Contains("raised"));
-        }
+        await WaitForEventAsync(events, e => e.SourceNodeId == "HighTemp" && e.AlarmType == "ALMD"
+            && e.Message.Contains("raised"));

        await drv.UnsubscribeAlarmsAsync(handle, CancellationToken.None);
        await drv.ShutdownAsync(CancellationToken.None);
@@ -135,15 +127,7 @@ public sealed class AbCipAlarmProjectionTests
    public async Task Resubscribe_Collapses_To_Single_Loop_No_Duplicate_Raise()
    {
        var factory = new FakeAbCipTagFactory();
-        var opts = new AbCipDriverOptions
-        {
-            Devices = [new AbCipDeviceOptions(Device)],
-            RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
-            EnableAlarmProjection = true,
-            AlarmPollInterval = TimeSpan.FromMilliseconds(20),
-            EnableDeclarationOnlyUdtGrouping = true,
-        };
-        var drv = new AbCipDriver(opts, "drv-1", factory);
+        var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory);
        await drv.InitializeAsync("{}", CancellationToken.None);

        var raises = new List<AlarmEventArgs>();
@@ -158,17 +142,18 @@ public sealed class AbCipAlarmProjectionTests
        await WaitForTagCreation(factory, "HighTemp");
        factory.Tags["HighTemp"].ValuesByOffset[0] = 0;   // InFaulted=false at offset 0
        factory.Tags["HighTemp"].ValuesByOffset[8] = 500; // Severity at offset 8
-        await Task.Delay(80); // let sub1's loop seed its "last-seen false" baseline
+        await WaitForMoreReadsAsync(factory, "HighTemp", 2); // sub1's loop has seeded its baseline

        // Second subscribe = the actor re-subscribing across a reconnect. Its loop reads the same
-        // shared HighTemp runtime; give it time to seed its own "last-seen false" baseline too.
+        // shared HighTemp runtime; let it poll (and so seed its own "last-seen false" baseline) too.
        var sub2 = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
-        await Task.Delay(80);
+        await WaitForMoreReadsAsync(factory, "HighTemp", 3);

        factory.Tags["HighTemp"].ValuesByOffset[0] = 1; // one 0->1 transition
-        // Wait past several 20ms poll intervals so any leaked second loop has ample time to also
-        // fire its duplicate raise before we assert.
-        await Task.Delay(250);
+        await WaitForEventAsync(raises, _ => true);
+        // Processed-count wait (#583), not a fixed delay: six more reads of the shared runtime after the raise,
+        // so a leaked second loop — which reads it too — has had polls in which to fire its duplicate.
+        await WaitForMoreReadsAsync(factory, "HighTemp", 6);

        await drv.UnsubscribeAlarmsAsync(sub1, CancellationToken.None);
        await drv.UnsubscribeAlarmsAsync(sub2, CancellationToken.None);
@@ -186,17 +171,7 @@ public sealed class AbCipAlarmProjectionTests
    public async Task Clear_Event_Fires_On_1_to_0_Transition()
    {
        var factory = new FakeAbCipTagFactory();
-        var opts = new AbCipDriverOptions
-        {
-            Devices = [new AbCipDeviceOptions(Device)],
-            RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
-            EnableAlarmProjection = true,
-            AlarmPollInterval = TimeSpan.FromMilliseconds(20),
-            // The ALMD projection here drives the parent-UDT runtime via offset-keyed values,
-            // so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003).
-            EnableDeclarationOnlyUdtGrouping = true,
-        };
-        var drv = new AbCipDriver(opts, "drv-1", factory);
+        var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory);
        await drv.InitializeAsync("{}", CancellationToken.None);

        var events = new List<AlarmEventArgs>();
@@ -207,16 +182,10 @@ public sealed class AbCipAlarmProjectionTests

        factory.Tags["HighTemp"].ValuesByOffset[0] = 1;
        factory.Tags["HighTemp"].ValuesByOffset[8] = 500;
-        await Task.Delay(80); // observe raise
+        await WaitForEventAsync(events, e => e.Message.Contains("raised"));

        factory.Tags["HighTemp"].ValuesByOffset[0] = 0;
-        await Task.Delay(200);
-
-        lock (events)
-        {
-            events.ShouldContain(e => e.Message.Contains("raised"));
-            events.ShouldContain(e => e.Message.Contains("cleared"));
-        }
+        await WaitForEventAsync(events, e => e.Message.Contains("cleared"));

        await drv.UnsubscribeAlarmsAsync(handle, CancellationToken.None);
        await drv.ShutdownAsync(CancellationToken.None);
@@ -227,33 +196,59 @@ public sealed class AbCipAlarmProjectionTests
    public async Task Unsubscribe_Stops_The_Poll_Loop()
    {
        var factory = new FakeAbCipTagFactory();
-        var opts = new AbCipDriverOptions
-        {
-            Devices = [new AbCipDeviceOptions(Device)],
-            RawTags = AbCipRawTags.From(AlmdTag("HighTemp")),
-            EnableAlarmProjection = true,
-            AlarmPollInterval = TimeSpan.FromMilliseconds(20),
-            // The ALMD projection here drives the parent-UDT runtime via offset-keyed values,
-            // so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003).
-            EnableDeclarationOnlyUdtGrouping = true,
-        };
-        var drv = new AbCipDriver(opts, "drv-1", factory);
+        var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory);
        await drv.InitializeAsync("{}", CancellationToken.None);

        var handle = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
        await WaitForTagCreation(factory, "HighTemp");
-        var preUnsubReadCount = factory.Tags["HighTemp"].ReadCount;
+        await WaitForMoreReadsAsync(factory, "HighTemp", 1);   // the loop is running

+        // Unsubscribe awaits the loop's exit, so no read can land after it returns.
        await drv.UnsubscribeAlarmsAsync(handle, CancellationToken.None);
-        await Task.Delay(100); // well past several poll intervals if the loop were still alive
+        var postUnsubReadCount = factory.Tags["HighTemp"].ReadCount;

-        var postDelayReadCount = factory.Tags["HighTemp"].ReadCount;
-        // Allow at most one straggler read between the unsubscribe-cancel + the loop exit.
-        (postDelayReadCount - preUnsubReadCount).ShouldBeLessThanOrEqualTo(1);
+        // Positive control (#583): an identical, still-subscribed driver polls three more times on the same
+        // interval — so the unsubscribed loop, had it survived, would have read again by now.
+        await PolledThreeTimesWithTheFlagOnAsync();
+        factory.Tags["HighTemp"].ReadCount.ShouldBe(postUnsubReadCount);

        await drv.ShutdownAsync(CancellationToken.None);
    }

+    /// <summary>A control driver with the projection ON polls its ALMD runtime three times.</summary>
+    private static async Task PolledThreeTimesWithTheFlagOnAsync()
+    {
+        var factory = new FakeAbCipTagFactory();
+        var control = new AbCipDriver(Options(enabled: true), "drv-control", factory);
+        await control.InitializeAsync("{}", CancellationToken.None);
+        var handle = await control.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
+        await WaitForTagCreation(factory, "HighTemp");
+        await WaitForMoreReadsAsync(factory, "HighTemp", 3);
+        await control.UnsubscribeAlarmsAsync(handle, CancellationToken.None);
+        await control.ShutdownAsync(CancellationToken.None);
+    }
+
+    /// <summary>Waits for <paramref name="more"/> further reads of the fake runtime (a processed-count wait).</summary>
+    private static async Task WaitForMoreReadsAsync(FakeAbCipTagFactory factory, string tagName, int more)
+    {
+        var tag = factory.Tags[tagName];
+        var target = tag.ReadCount + more;
+        var deadline = DateTime.UtcNow.AddSeconds(5);
+        while (tag.ReadCount < target && DateTime.UtcNow < deadline) await Task.Delay(5);
+        tag.ReadCount.ShouldBeGreaterThanOrEqualTo(target, $"'{tagName}' should have been read {more} more times");
+    }
+
+    private static async Task WaitForEventAsync(List<AlarmEventArgs> events, Func<AlarmEventArgs, bool> match)
+    {
+        var deadline = DateTime.UtcNow.AddSeconds(5);
+        while (DateTime.UtcNow < deadline)
+        {
+            lock (events) if (events.Any(match)) return;
+            await Task.Delay(5);
+        }
+        throw new TimeoutException("The expected alarm event was never raised.");
+    }
+
    private static async Task WaitForTagCreation(FakeAbCipTagFactory factory, string tagName)
    {
        var deadline = DateTime.UtcNow.AddSeconds(2);
@@ -152,8 +152,11 @@ public sealed class AbCipDriverCodeReviewRegressionTests
        }, "drv-1", factory);

        await drv.InitializeAsync("{}", CancellationToken.None);
-        // Give the probe loop a moment to actually start spinning.
-        await Task.Delay(60);
+        // Wait until the probe loop is actually spinning (#583: a processed count, not a fixed moment).
+        var deadline = DateTime.UtcNow.AddSeconds(5);
+        while (!(factory.Tags.TryGetValue("ProbeTag", out var probe) && probe.ReadCount >= 2) && DateTime.UtcNow < deadline)
+            await Task.Delay(5, TestContext.Current.CancellationToken);
+        factory.Tags["ProbeTag"].ReadCount.ShouldBeGreaterThanOrEqualTo(2);

        // Must complete cleanly — no ObjectDisposedException from a loop racing a disposed CTS.
        await Should.NotThrowAsync(() => drv.ShutdownAsync(CancellationToken.None));
@@ -100,11 +100,36 @@ public sealed class AbCipHostProbeTests
        drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);

        await drv.InitializeAsync("{}", CancellationToken.None);
-        await Task.Delay(300);

+        // #583: structural, not a fixed wait — no probe loop was ever started, so no transition can follow.
+        drv.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeTask.ShouldBeNull();
+        factory.Tags.ShouldBeEmpty();
        transitions.ShouldBeEmpty();
        drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown);
        await drv.ShutdownAsync(CancellationToken.None);
+        await ProbeLoopIsObservableAsync();
+    }
+
+    /// <summary>
+    ///     Positive control for the "no probe loop" assertions above: the same observation (the device's
+    ///     <c>ProbeTask</c>) on a driver whose probe IS configured is non-null, and that loop does produce a
+    ///     transition — so a null <c>ProbeTask</c> really means "no loop", not "a loop the hook cannot see".
+    /// </summary>
+    private static async Task ProbeLoopIsObservableAsync()
+    {
+        var factory = new FakeAbCipTagFactory { Customise = p => new FakeAbCipTag(p) { Status = 0 } };
+        var transitions = new ConcurrentQueue<HostStatusChangedEventArgs>();
+        var control = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")],
+            Probe = new AbCipProbeOptions { Enabled = true, ProbeTagPath = "@raw_cpu_type", Interval = TimeSpan.FromMilliseconds(100) },
+        }, "drv-control", factory);
+        control.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);
+        await control.InitializeAsync("{}", CancellationToken.None);
+        (control.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeTask is not null).ShouldBeTrue();
+        await WaitForAsync(() => !transitions.IsEmpty, TimeSpan.FromSeconds(2));
+        transitions.ShouldNotBeEmpty();
+        await control.ShutdownAsync(CancellationToken.None);
    }

    /// <summary>Verifies that the probe is skipped when ProbeTagPath is null.</summary>
@@ -118,10 +143,11 @@ public sealed class AbCipHostProbeTests
        }, "drv-1");

        await drv.InitializeAsync("{}", CancellationToken.None);
-        await Task.Delay(200);

+        drv.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeTask.ShouldBeNull();   // #583: no loop, so no transition
        drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown);
        await drv.ShutdownAsync(CancellationToken.None);
+        await ProbeLoopIsObservableAsync();
    }

    /// <summary>Verifies that the probe loops across multiple devices independently.</summary>
@@ -65,8 +65,12 @@ public sealed class AbCipLoggingTests

        await drv.InitializeAsync("{}", CancellationToken.None);

-        // Give the probe loop a couple of ticks to log.
-        await Task.Delay(200);
+        // Wait (bounded) for the entry itself rather than a fixed couple of ticks (#583).
+        var deadline = DateTime.UtcNow.AddSeconds(5);
+        while (!logger.Entries.Any(e => e.Message.Contains("probe", StringComparison.OrdinalIgnoreCase)
+                                        || (e.Exception?.Message.Contains("simulated probe init failure") ?? false))
+               && DateTime.UtcNow < deadline)
+            await Task.Delay(5, TestContext.Current.CancellationToken);

        await drv.ShutdownAsync(CancellationToken.None);

@@ -186,8 +190,10 @@ public sealed class AbCipLoggingTests
    /// <summary>Test logger that captures all log entries.</summary>
    internal sealed class CapturingLogger<T> : ILogger<T>
    {
-        /// <summary>Gets the captured log entries.</summary>
-        public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = new();
+        private readonly List<(LogLevel Level, string Message, Exception? Exception)> _entries = new();
+
+        /// <summary>Gets a snapshot of the captured log entries (the probe loop logs from another thread).</summary>
+        public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get { lock (_entries) return [.. _entries]; } }

        /// <summary>Begins a scope (stub implementation).</summary>
        /// <typeparam name="TState">The type of the scope state.</typeparam>
@@ -206,7 +212,7 @@ public sealed class AbCipLoggingTests
        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
            Func<TState, Exception?, string> formatter)
        {
-            Entries.Add((logLevel, formatter(state, exception), exception));
+            lock (_entries) _entries.Add((logLevel, formatter(state, exception), exception));
        }

        private sealed class NullScope : IDisposable
@@ -67,7 +67,8 @@ public sealed class AbCipSubscriptionTests
        drv.OnDataChange += (_, e) => events.Enqueue(e);

        var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
-        await Task.Delay(500);
+        // Processed-count wait (#583): five polls have read the unchanged value, not "500 ms went by".
+        await WaitForReadsAsync(factory, "Speed", 5);
        await drv.UnsubscribeAsync(handle, CancellationToken.None);

        events.Count.ShouldBe(1);
@@ -101,22 +102,30 @@ public sealed class AbCipSubscriptionTests
    public async Task Unsubscribe_halts_polling()
    {
        var (drv, factory) = NewDriver(
-            new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt));
+            new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt),
+            new AbCipTagDefinition("Temp", "ab://10.0.0.5/1,0", "Temp", AbCipDataType.DInt));
        await drv.InitializeAsync("{}", CancellationToken.None);
        var tagRef = new FakeAbCipTag(new AbCipTagCreateParams("10.0.0.5", 44818, "1,0", "controllogix", "Speed", TimeSpan.FromSeconds(2))) { Value = 1 };
-        factory.Customise = _ => tagRef;
+        factory.Customise = p => p.TagName == "Speed" ? tagRef : new FakeAbCipTag(p) { Value = 1 };

        var events = new ConcurrentQueue<DataChangeEventArgs>();
        drv.OnDataChange += (_, e) => events.Enqueue(e);

        var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
-        await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1));
+        var controlHandle = await drv.SubscribeAsync(["Temp"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
+        await WaitForAsync(() => events.Any(e => e.FullReference == "Speed"), TimeSpan.FromSeconds(1));
        await drv.UnsubscribeAsync(handle, CancellationToken.None);

-        var afterUnsub = events.Count;
+        var afterUnsub = events.Count(e => e.FullReference == "Speed");
+        var readsAfterUnsub = tagRef.ReadCount;
        tagRef.Value = 999;
-        await Task.Delay(400);
-        events.Count.ShouldBe(afterUnsub);
+        // Positive control (#583): the still-subscribed sibling, on the same interval, polls three more times —
+        // so Speed's loop, had it survived the unsubscribe, has had three chances to read the change.
+        await WaitForReadsAsync(factory, "Temp", factory.Tags["Temp"].ReadCount + 3);
+
+        events.Count(e => e.FullReference == "Speed").ShouldBe(afterUnsub);
+        tagRef.ReadCount.ShouldBe(readsAfterUnsub);
+        await drv.UnsubscribeAsync(controlHandle, CancellationToken.None);
    }

    /// <summary>Verifies that polling intervals below 100ms are floored to the minimum.</summary>
@@ -132,12 +141,17 @@ public sealed class AbCipSubscriptionTests
        drv.OnDataChange += (_, e) => events.Enqueue(e);

        var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(5), CancellationToken.None);
-        await Task.Delay(300);
+        // Processed-count wait (#583): three polls, then measure the spacing they actually ran at.
+        var started = System.Diagnostics.Stopwatch.StartNew();
+        await WaitForReadsAsync(factory, "Speed", 3);
+        var elapsed = started.Elapsed;
        await drv.UnsubscribeAsync(handle, CancellationToken.None);

-        // Value is stable → only the initial-data push fires; the 100 ms floor keeps polls sparse enough
-        // that no extra event is produced against a stable value.
+        // Value is stable → only the initial-data push fires.
        events.Count.ShouldBe(1);
+        // The floor: the initial read is immediate, so the two later ones took at least two 100 ms intervals.
+        // At the requested 5 ms they would have taken ~10 ms.
+        elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(180));
    }

    /// <summary>Verifies that ShutdownAsync cancels all active subscriptions.</summary>
@@ -154,11 +168,25 @@ public sealed class AbCipSubscriptionTests

        _ = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
        await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1));
+        var speed = factory.Tags["Speed"];
        await drv.ShutdownAsync(CancellationToken.None);

        var afterShutdown = events.Count;
-        await Task.Delay(300);
+        var readsAfterShutdown = speed.ReadCount;
+        speed.Value = 2;   // a change any surviving loop would raise
+
+        // Positive control (#583): a second, independent driver polling on the same interval reads three more
+        // times — so the shut-down driver's loop, had it survived, has had three chances to raise the change.
+        var (control, controlFactory) = NewDriver(
+            new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt));
+        await control.InitializeAsync("{}", CancellationToken.None);
+        controlFactory.Customise = p => new FakeAbCipTag(p) { Value = 1 };
+        _ = await control.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
+        await WaitForReadsAsync(controlFactory, "Speed", 3);
+        await control.ShutdownAsync(CancellationToken.None);
+
        events.Count.ShouldBe(afterShutdown);
+        speed.ReadCount.ShouldBe(readsAfterShutdown);
    }

    /// <summary>Verifies that subscriptions on UDT members use the synthesized full reference.</summary>
@@ -195,4 +223,12 @@ public sealed class AbCipSubscriptionTests
        while (!condition() && DateTime.UtcNow < deadline)
            await Task.Delay(20);
    }
+
+    /// <summary>Waits until the fake runtime for <paramref name="tag"/> has served at least <paramref name="reads"/>
+    ///     reads (a processed-count wait), failing — never passing vacuously — if it does not get there.</summary>
+    private static async Task WaitForReadsAsync(FakeAbCipTagFactory factory, string tag, int reads)
+    {
+        await WaitForAsync(() => factory.Tags.TryGetValue(tag, out var t) && t.ReadCount >= reads, TimeSpan.FromSeconds(5));
+        factory.Tags[tag].ReadCount.ShouldBeGreaterThanOrEqualTo(reads, $"'{tag}' should have been polled {reads} times");
+    }
 }
@@ -25,7 +25,8 @@ internal class FakeAbCipTag : IAbCipTagRuntime
    /// <summary>Gets the count of <see cref="InitializeAsync"/> invocations.</summary>
    public int InitializeCount { get; private set; }
    /// <summary>Gets the count of <see cref="ReadAsync"/> invocations.</summary>
-    public int ReadCount { get; private set; }
+    public int ReadCount => Volatile.Read(ref _readCount);
+    private int _readCount;   // interlocked: poll loops read it from another thread (#583 processed-count waits)
    /// <summary>Gets the count of <see cref="WriteAsync"/> invocations.</summary>
    public int WriteCount { get; private set; }
    /// <summary>Gets a value indicating whether the tag has been disposed.</summary>
@@ -48,7 +49,7 @@ internal class FakeAbCipTag : IAbCipTagRuntime
    /// <param name="cancellationToken">The cancellation token for the operation.</param>
    public virtual Task ReadAsync(CancellationToken cancellationToken)
    {
-        ReadCount++;
+        Interlocked.Increment(ref _readCount);
        if (ThrowOnRead) throw Exception ?? new InvalidOperationException("fake read failure");
        return Task.CompletedTask;
    }
@@ -72,7 +72,9 @@ public sealed class NativeAbCipLeaseReleaseTests
        // the first one's Forward Close is in flight (review finding 22) — the pool holds it behind the close.
        var second = (NativeAbCipTagRuntime)factory.Create(Params(target.Port));
        var init = second.InitializeAsync(Ct);
-        await Task.Delay(300, Ct);
+        // Positive control (#583): the second open has reached the pool's predecessor wait and parked there, so
+        // the missing Forward Open below is that wait holding it — not the open simply not having got that far.
+        await FakeCipTarget.Until(() => pool.Diagnostics.PredecessorWaits == 1, Guard, "the re-create to park behind the close");
        target.ForwardOpens.ShouldBe(1);
        init.IsCompleted.ShouldBeFalse();

@@ -71,11 +71,17 @@ public sealed class AbLegacyCapabilityTests
    {
        var tagRef = new FakeAbLegacyTag(
            new AbLegacyTagCreateParams("10.0.0.5", 44818, "1,0", "slc500", "N7:0", TimeSpan.FromSeconds(2))) { Value = 1 };
-        var factory = new FakeAbLegacyTagFactory { Customise = _ => tagRef };
+        var control = new FakeAbLegacyTag(
+            new AbLegacyTagCreateParams("10.0.0.5", 44818, "1,0", "slc500", "N7:1", TimeSpan.FromSeconds(2))) { Value = 1 };
+        var factory = new FakeAbLegacyTagFactory { Customise = p => p.TagName == "N7:1" ? control : tagRef };
        var drv = new AbLegacyDriver(new AbLegacyDriverOptions
        {
            Devices = [new AbLegacyDeviceOptions("ab://10.0.0.5/1,0")],
-            RawTags = AbLegacyRawTags.Entries([new AbLegacyTagDefinition("X", "ab://10.0.0.5/1,0", "N7:0", AbLegacyDataType.Int)]),
+            RawTags = AbLegacyRawTags.Entries(
+            [
+                new AbLegacyTagDefinition("X", "ab://10.0.0.5/1,0", "N7:0", AbLegacyDataType.Int),
+                new AbLegacyTagDefinition("Y", "ab://10.0.0.5/1,0", "N7:1", AbLegacyDataType.Int),
+            ]),
            Probe = new AbLegacyProbeOptions { Enabled = false },
        }, "drv-1", factory);
        await drv.InitializeAsync("{}", CancellationToken.None);
@@ -84,13 +90,22 @@ public sealed class AbLegacyCapabilityTests
        drv.OnDataChange += (_, e) => events.Enqueue(e);

        var handle = await drv.SubscribeAsync(["X"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
-        await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1));
+        var controlHandle = await drv.SubscribeAsync(["Y"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
+        await WaitForAsync(() => events.Count(e => e.FullReference == "X") >= 1, TimeSpan.FromSeconds(1));
        await drv.UnsubscribeAsync(handle, CancellationToken.None);

-        var afterUnsub = events.Count;
+        var afterUnsub = events.Count(e => e.FullReference == "X");
+        var readsAfterUnsub = tagRef.ReadCount;
        tagRef.Value = 999;
-        await Task.Delay(300, TestContext.Current.CancellationToken);
-        events.Count.ShouldBe(afterUnsub);
+        // Positive control (#583): the still-subscribed sibling, on the same interval, polls three more times —
+        // so X's loop, had it survived, has had three chances to read the change and raise it.
+        var controlReads = control.ReadCount;
+        await WaitForAsync(() => control.ReadCount >= controlReads + 3, TimeSpan.FromSeconds(3));
+        control.ReadCount.ShouldBeGreaterThanOrEqualTo(controlReads + 3);
+
+        events.Count(e => e.FullReference == "X").ShouldBe(afterUnsub);
+        tagRef.ReadCount.ShouldBe(readsAfterUnsub);
+        await drv.UnsubscribeAsync(controlHandle, CancellationToken.None);
    }

    // ---- IHostConnectivityProbe ----
@@ -171,10 +186,30 @@ public sealed class AbLegacyCapabilityTests
            Probe = new AbLegacyProbeOptions { Enabled = true, ProbeAddress = null },
        }, "drv-1");
        await drv.InitializeAsync("{}", CancellationToken.None);
-        await Task.Delay(200, TestContext.Current.CancellationToken);

+        // #583: structural, not a fixed wait — no probe loop was ever started, so no transition can follow.
+        drv.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeCts.ShouldBeNull();
        drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown);
        await drv.ShutdownAsync(CancellationToken.None);
+
+        // Positive control: the same observation on a driver whose probe IS configured sees the loop, and that loop
+        // produces a transition — so a null ProbeCts really means "no loop".
+        var transitions = new ConcurrentQueue<HostStatusChangedEventArgs>();
+        var control = new AbLegacyDriver(new AbLegacyDriverOptions
+        {
+            Devices = [new AbLegacyDeviceOptions("ab://10.0.0.5/1,0")],
+            Probe = new AbLegacyProbeOptions
+            {
+                Enabled = true, Interval = TimeSpan.FromMilliseconds(100),
+                Timeout = TimeSpan.FromMilliseconds(50), ProbeAddress = "S:0",
+            },
+        }, "drv-control", new FakeAbLegacyTagFactory());
+        control.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);
+        await control.InitializeAsync("{}", CancellationToken.None);
+        control.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeCts.ShouldNotBeNull();
+        await WaitForAsync(() => !transitions.IsEmpty, TimeSpan.FromSeconds(2));
+        transitions.ShouldNotBeEmpty();
+        await control.ShutdownAsync(CancellationToken.None);
    }

    // ---- IPerCallHostResolver ----
@@ -43,7 +43,8 @@ internal class FakeAbLegacyTag : IAbLegacyTagRuntime
    public int InitializeCount { get; private set; }

    /// <summary>Gets the count of read calls.</summary>
-    public int ReadCount { get; private set; }
+    public int ReadCount => Volatile.Read(ref _readCount);
+    private int _readCount;   // interlocked: poll loops read it from another thread (#583 processed-count waits)

    /// <summary>Gets the count of write calls.</summary>
    public int WriteCount { get; private set; }
@@ -70,7 +71,7 @@ internal class FakeAbLegacyTag : IAbLegacyTagRuntime
    /// <returns>A task representing the asynchronous operation.</returns>
    public virtual Task ReadAsync(CancellationToken ct)
    {
-        ReadCount++;
+        Interlocked.Increment(ref _readCount);
        if (ThrowOnRead) throw Exception ?? new InvalidOperationException();
        return Task.CompletedTask;
    }