fix(717): Warm crash survivor stays Active through auto-down; verdict reads its own events

Two defects on the crash-failover path, found by the T13 live gate on an isolated docker pair:

1. When auto-down removed the unreachable Primary (~15 s after the survivor promoted), the
   snapshot's Primary left the unreachable set, so (3') failed and the survivor demoted to
   Standby (BadOutOfService, ServiceLevel 50) for ~15 s, until its own singleton published
   after the hand-over retries. A Primary that is Down, or that this node saw removed, now
   counts as gone (WarmRoleAuthority.SnapshotPrimaryGone). A member that is merely
   Leaving/Exiting, or that this node never saw, is not.

2. The verdict read Cluster.State / SelfMember (Akka's ClusterReadView, a separate subscriber
   to the same events), which can lag the event being handled. Live, the survivor stayed
   Standby 4.3 s after its own UnreachableMember event (promotion waited for the next health
   tick). The Warm verdict now uses a membership view built from this actor's own
   member/reachability subscription, falling back to the read view only before the first event.

Tests: WarmRoleAuthorityTests.A_snapshot_Primary_that_auto_down_removed_is_still_gone_and_the_survivor_stays_Active,
OpcUaPublishActorWarmTests.The_crash_survivor_promotes_on_its_own_unreachable_event_and_stays_Active_through_the_removal
(synthetic events the read view never sees). Mutations: read-view-only verdict and the removed
arm switched off each fail the actor test. Runtime.Tests 1058 passed, 3 skipped.

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

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs

@@ -540,9 +540,21 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         Receive<SubscribeAck>(_ => { /* PubSub ack */ });
         // Gitea #717 spec §4.1 / owner answer Q2: under Warm the verdict re-derives on membership and reachability, not
         // only on snapshots. Nothing subscribes these under Unset/None.
-        Receive<ClusterEvent.IMemberEvent>(_ => RecomputeServiceLevel());
-        Receive<ClusterEvent.IReachabilityEvent>(_ => RecomputeServiceLevel());
-        Receive<ClusterEvent.CurrentClusterState>(_ => RecomputeServiceLevel());
+        Receive<ClusterEvent.IMemberEvent>(e =>
+        {
+            TrackMember(e);
+            RecomputeServiceLevel();
+        });
+        Receive<ClusterEvent.IReachabilityEvent>(e =>
+        {
+            TrackReachability(e);
+            RecomputeServiceLevel();
+        });
+        Receive<ClusterEvent.CurrentClusterState>(state =>
+        {
+            ResetWarmView(state);
+            RecomputeServiceLevel();
+        });
         Receive<WarmRetry>(_ => RecomputeServiceLevel());
     }
 
@@ -1730,6 +1742,61 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     private readonly SnapshotRevisionGate _revisionGate = new();
     /// <summary>The latest ACCEPTED revisioned snapshot — the only one the Warm verdict reads (spec §4.2).</summary>
     private RedundancyStateChanged? _acceptedSnapshot;
+
+    // Gitea #717 T13 live gate: the Warm verdict reads membership from THIS actor's own event subscription, not from
+    // Cluster.State / Cluster.SelfMember. Those come from Akka's ClusterReadView, a separate subscriber to the same
+    // events, so when this actor handles an event the read view may not show it yet. Live, that held a crash survivor
+    // Standby for 4.3 s after its own UnreachableMember event (promotion waited for the next health tick), and let a
+    // gracefully leaving Primary keep serving Good ~240 ms after the survivor was already Active.
+    private readonly Dictionary<Address, Member> _warmMembers = [];
+    private readonly HashSet<Address> _warmUnreachable = [];
+
+    // Gitea #717 T13: nodes seen removed from the cluster (and not back since). A snapshot naming one is as gone as an
+    // unreachable one, so the survivor of a crash stays Active through auto-down's removal (WarmRoleAuthority.SnapshotPrimaryGone).
+    private readonly HashSet<NodeId> _removedMembers = [];
+
+    private void TrackMember(ClusterEvent.IMemberEvent e)
+    {
+        var node = WarmRoleAuthority.ToNodeId(e.Member);
+        if (e is ClusterEvent.MemberRemoved)
+        {
+            _warmMembers.Remove(e.Member.Address);
+            _warmUnreachable.Remove(e.Member.Address);
+            _removedMembers.Add(node);
+            return;
+        }
+
+        _warmMembers[e.Member.Address] = e.Member;
+        if (e is ClusterEvent.MemberJoined or ClusterEvent.MemberUp) _removedMembers.Remove(node);
+    }
+
+    private void TrackReachability(ClusterEvent.IReachabilityEvent e)
+    {
+        if (e is ClusterEvent.UnreachableMember u) _warmUnreachable.Add(u.Member.Address);
+        else if (e is ClusterEvent.ReachableMember r) _warmUnreachable.Remove(r.Member.Address);
+    }
+
+    private void ResetWarmView(ClusterEvent.CurrentClusterState state)
+    {
+        _warmMembers.Clear();
+        _warmUnreachable.Clear();
+        foreach (var m in state.Members) _warmMembers[m.Address] = m;
+        foreach (var m in state.Unreachable) _warmUnreachable.Add(m.Address);
+    }
+
+    /// <summary>The Warm view's own member status, falling back to the read view before the first event arrives.</summary>
+    private MemberStatus WarmSelfStatus()
+    {
+        try
+        {
+            return _warmMembers.TryGetValue(_cluster.SelfAddress, out var self) ? self.Status : SafeSelfStatus();
+        }
+        catch (Exception ex)
+        {
+            _log.Debug(ex, "OpcUaPublish: self address unavailable for the Warm view; using the read view");
+            return SafeSelfStatus();
+        }
+    }
     /// <summary>The role the data plane is in. A Warm node manager boots Standby, so this starts Standby.</summary>
     private WarmRole _warmRole = WarmRole.Standby;
     /// <summary>A promotion whose Active ServiceLevel publish failed (spec §4.7): the data plane is Active and the
@@ -1756,7 +1823,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         var self = _localNode!.Value;
         var entry = _lastSnapshot!.Nodes.FirstOrDefault(n => n.NodeId == self);
         var now = _utcNow();
-        var memberState = SafeSelfStatus();
+        var memberState = WarmSelfStatus();
         var listening = EndpointListening();
 
         bool dbReachable;
@@ -1791,11 +1858,23 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         var snapshotPrimary = _acceptedSnapshot?.Nodes.FirstOrDefault(n => n.Role == RedundancyRole.Primary)?.NodeId;
         try
         {
-            var state = _cluster.State;
-            var unreachable = state.Unreachable.ToHashSet();
-            oldest = WarmRoleAuthority.OldestUpMember(state.Members, "driver");
-            oldestReachable = WarmRoleAuthority.OldestUpMember(state.Members, "driver", unreachable);
-            primaryUnreachable = snapshotPrimary is { } sp && unreachable.Any(m => WarmRoleAuthority.ToNodeId(m) == sp);
+            IReadOnlyCollection<Member> members;
+            HashSet<Member> unreachable;
+            if (_warmMembers.Count > 0)
+            {
+                members = _warmMembers.Values;
+                unreachable = _warmMembers.Values.Where(m => _warmUnreachable.Contains(m.Address)).ToHashSet();
+            }
+            else
+            {
+                var state = _cluster.State;
+                members = state.Members;
+                unreachable = state.Unreachable.ToHashSet();
+            }
+
+            oldest = WarmRoleAuthority.OldestUpMember(members, "driver");
+            oldestReachable = WarmRoleAuthority.OldestUpMember(members, "driver", unreachable);
+            primaryUnreachable = WarmRoleAuthority.SnapshotPrimaryGone(snapshotPrimary, members, unreachable, _removedMembers);
         }
         catch (Exception ex)
         {

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/WarmRoleAuthority.cs

@@ -102,6 +102,36 @@ public static class WarmRoleAuthority
         return oldest is null ? null : ToNodeId(oldest);
     }
 
+    /// <summary>
+    ///     Whether the snapshot's Primary is gone from this node's view, for <see cref="WarmRoleInputs.SnapshotPrimaryUnreachable"/>:
+    ///     unreachable, marked <c>Down</c>, or removed from the cluster after this node saw it as a member.
+    /// </summary>
+    /// <remarks>
+    ///     The "removed" arm is the end of the same crash path (T13 live gate, 2026-11-13). Auto-down removes the
+    ///     unreachable Primary about 15 s after the survivor promoted, so it leaves the unreachable set. But the
+    ///     survivor's own singleton does not publish a snapshot naming it until the hand-over retries time out, about
+    ///     15 s later. Without this arm the survivor demoted for that whole window (BadOutOfService, ServiceLevel 50)
+    ///     and then promoted again. A member that is merely Leaving or Exiting is still present and reachable, so a
+    ///     graceful hand-over is unaffected, and a node this one never saw as a member is not assumed gone.
+    /// </remarks>
+    /// <param name="snapshotPrimary">The Primary named by the accepted snapshot, or null.</param>
+    /// <param name="members">The current members (this node's own cluster state).</param>
+    /// <param name="unreachable">The members this node sees as unreachable.</param>
+    /// <param name="removed">Nodes this node has seen removed from the cluster (and not rejoin since).</param>
+    /// <returns>True when a named Primary is unreachable, Down, or removed.</returns>
+    public static bool SnapshotPrimaryGone(NodeId? snapshotPrimary, IEnumerable<Member> members, IReadOnlySet<Member> unreachable,
+        IReadOnlySet<NodeId> removed)
+    {
+        if (snapshotPrimary is not { } sp) return false;
+        foreach (var m in members)
+        {
+            if (ToNodeId(m) != sp) continue;
+            return m.Status is MemberStatus.Down or MemberStatus.Removed || unreachable.Contains(m);
+        }
+
+        return removed.Contains(sp);
+    }
+
     /// <summary>The canonical <c>host:port</c> id of a member (the RedundancyStateActor format).</summary>
     /// <param name="m">The member.</param>
     /// <returns>The node id.</returns>

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorWarmTests.cs

@@ -157,6 +157,68 @@ public sealed class OpcUaPublishActorWarmTests : RuntimeActorTestBase
         AssertNoActiveLevelOverProjectedData();
     }
 
+    /// <summary>
+    ///     T13 live gate (2026-11-13), two findings on the crash path, driven here with synthetic cluster events that
+    ///     Akka's read view (Cluster.State) never sees:
+    ///     (1) the verdict reacts to the event it is handling, not to the read view, which can lag it (live: the survivor
+    ///     stayed Standby 4.3 s after its own UnreachableMember event);
+    ///     (2) when auto-down removes the unreachable Primary, the survivor stays Active (live: it demoted for ~15 s until
+    ///     its own singleton published).
+    /// </summary>
+    [Fact]
+    public void The_crash_survivor_promotes_on_its_own_unreachable_event_and_stays_Active_through_the_removal()
+    {
+        var actor = Start(new WarmRedundancySettings(OpcUaRedundancyMode.Warm));
+        var other = MemberAt(Other);
+        actor.Tell(new Akka.Cluster.ClusterEvent.MemberUp(other));
+        actor.Tell(Snapshot(Other));
+        AwaitLogEndsWith("SL:90");
+
+        actor.Tell(new Akka.Cluster.ClusterEvent.UnreachableMember(other));
+        AwaitLogEndsWith(Promotion);
+
+        actor.Tell(new Akka.Cluster.ClusterEvent.MemberRemoved(MemberAt(Other, Akka.Cluster.MemberStatus.Removed), Akka.Cluster.MemberStatus.Down));
+        actor.Tell(Snapshot(Other, upNumber: 1)); // the same accepted revision line, still naming the departed Primary
+        ExpectNoMsg(TimeSpan.FromMilliseconds(500), TestContext.Current.CancellationToken);
+        Steps.ShouldNotContain("delivery:True", "a Primary removed by auto-down is still gone; the survivor stays Active");
+
+        // Positive control: the departed Primary comes back Up and reachable, so the snapshot naming it demotes.
+        actor.Tell(new Akka.Cluster.ClusterEvent.MemberUp(other));
+        actor.Tell(new Akka.Cluster.ClusterEvent.ReachableMember(other));
+        actor.Tell(Snapshot(Other, upNumber: 1));
+        AwaitLogEndsWith(Demotion);
+    }
+
+    private static int _memberUid = 1000;
+
+    // Akka's Member.Create factories are internal; same reflection idiom as DriverHostActorHistorizationRoleViewTests.
+    private static readonly System.Reflection.MethodInfo MemberCreate = typeof(Akka.Cluster.Member).GetMethod(
+            "Create",
+            System.Reflection.BindingFlags.Static | System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Public,
+            binder: null,
+            types:
+            [
+                typeof(Akka.Cluster.UniqueAddress), typeof(int), typeof(Akka.Cluster.MemberStatus),
+                typeof(System.Collections.Immutable.ImmutableHashSet<string>), typeof(Akka.Util.AppVersion),
+            ],
+            modifiers: null)
+        ?? throw new InvalidOperationException("Akka.Cluster.Member.Create(5-arg) not found; Akka API changed");
+
+    private static Akka.Cluster.Member MemberAt(NodeId node, Akka.Cluster.MemberStatus status = Akka.Cluster.MemberStatus.Up)
+    {
+        var text = node.ToString();
+        var colon = text.LastIndexOf(':');
+        return (Akka.Cluster.Member)MemberCreate.Invoke(null,
+        [
+            new Akka.Cluster.UniqueAddress(
+                new Address("akka.tcp", "otopcua", text[..colon], int.Parse(text[(colon + 1)..])), 7),
+            Interlocked.Increment(ref _memberUid),
+            status,
+            System.Collections.Immutable.ImmutableHashSet.Create("driver"),
+            Akka.Util.AppVersion.Zero,
+        ])!;
+    }
+
     /// <summary>The default-off pin at the actor: Unset never touches the projection and keeps today's levels.</summary>
     [Fact]
     public void Unset_never_drives_the_projection_and_keeps_todays_levels()

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Redundancy/WarmRoleAuthorityTests.cs

@@ -1,3 +1,8 @@
+using System.Collections.Immutable;
+using System.Reflection;
+using Akka.Actor;
+using Akka.Cluster;
+using Akka.Util;
 using Shouldly;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
@@ -61,6 +66,70 @@ public sealed class WarmRoleAuthorityTests
             .ShouldBe(WarmRole.Standby, "a node that never accepted a snapshot stays Standby");
     }
 
+    /// <summary>
+    ///     T13 live gate (2026-11-13): auto-down removes the crashed Primary ~15 s after the survivor promoted, while the
+    ///     survivor's own singleton has not yet published a snapshot naming it. A Primary that is Down or no longer a
+    ///     member is as gone as an unreachable one, so the survivor stays Active through the removal. A Primary that is
+    ///     present, reachable and merely Leaving/Exiting (a graceful hand-over) is not gone.
+    /// </summary>
+    [Fact]
+    public void A_snapshot_Primary_that_auto_down_removed_is_still_gone_and_the_survivor_stays_Active()
+    {
+        var self = MemberAt(Self);
+        var peer = MemberAt(Peer);
+        var none = new HashSet<Member>();
+        var seenRemoved = new HashSet<NodeId> { Peer };
+        var noneRemoved = new HashSet<NodeId>();
+
+        WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, peer], new HashSet<Member> { peer }, noneRemoved).ShouldBeTrue("unreachable");
+        WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, MemberAt(Peer, MemberStatus.Down)], none, noneRemoved).ShouldBeTrue("downed");
+        WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self], none, seenRemoved).ShouldBeTrue("removed by auto-down");
+
+        WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self], none, noneRemoved).ShouldBeFalse("never seen as a member: not assumed gone");
+        WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, peer], none, seenRemoved).ShouldBeFalse("back as a reachable Up member");
+        WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, peer], none, noneRemoved).ShouldBeFalse("a reachable Up Primary is never overridden");
+        WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, MemberAt(Peer, MemberStatus.Leaving)], none, noneRemoved).ShouldBeFalse("graceful leave");
+        WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, MemberAt(Peer, MemberStatus.Exiting)], none, noneRemoved).ShouldBeFalse("graceful exit");
+        WarmRoleAuthority.SnapshotPrimaryGone(null, [self], none, seenRemoved).ShouldBeFalse("no snapshot Primary");
+
+        // The live sequence on the survivor: promoted while the Primary was unreachable, then the Primary is removed.
+        var removed = In(oldest: Self, oldestReachable: Self, snapshotPrimary: Peer,
+            primaryUnreachable: WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self], none, seenRemoved));
+        WarmRoleAuthority.Evaluate(removed).Role.ShouldBe(WarmRole.Active);
+        WarmRoleAuthority.Evaluate(removed with { PromoteWhilePrimaryUnreachable = false }).Role.ShouldBe(WarmRole.Standby);
+    }
+
+    private static int _memberUid;
+
+    // Akka's Member.Create factories are internal; same reflection idiom as DriverHostActorHistorizationRoleViewTests.
+    private static readonly MethodInfo MemberCreate = typeof(Member).GetMethod(
+            "Create",
+            BindingFlags.Static | BindingFlags.NonPublic | BindingFlags.Public,
+            binder: null,
+            types:
+            [
+                typeof(UniqueAddress), typeof(int), typeof(MemberStatus),
+                typeof(ImmutableHashSet<string>), typeof(AppVersion),
+            ],
+            modifiers: null)
+        ?? throw new InvalidOperationException("Akka.Cluster.Member.Create(5-arg) not found; Akka API changed");
+
+    private static Member MemberAt(NodeId node, MemberStatus status = MemberStatus.Up)
+    {
+        var text = node.ToString();
+        var colon = text.LastIndexOf(':');
+        return (Member)MemberCreate.Invoke(null,
+        [
+            new UniqueAddress(
+                new Address("akka.tcp", "otopcua", text[..colon], int.Parse(text[(colon + 1)..])),
+                Interlocked.Increment(ref _memberUid)),
+            1,
+            status,
+            ImmutableHashSet.Create("driver"),
+            AppVersion.Zero,
+        ])!;
+    }
+
     [Fact]
     public void A_late_bind_replays_the_retained_state_but_not_the_actions()
     {