fix(cluster): don't self-form while a seed peer is reachable — the fallback islanded a restarting node

The live gate for the manual-failover control caught the self-form fallback
forming a SECOND cluster. A node bounced by a failover restarted, received
InitJoinAck from its live peer — the join was in flight and healthy — but did
not get the Welcome inside the 10s window, because the peer's ring still held
the node's previous incarnation (Exiting -> Down -> Removed). The fallback
fired on the timer and the node islanded itself until an operator restarted it.

The original design assumed Cluster.Join(SelfAddress) would be ignored
mid-handshake. It is not — it wins. And since manual failover deliberately
produces that restart, every failover could island the node it bounced.

Before self-forming, TCP-probe the other seed addresses; a reachable peer means
wait another window instead. That is also what the fallback claims to detect:
'no seed answered InitJoin (peer down at boot)'.

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

src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterBootstrapFallback.cs

@@ -20,11 +20,22 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster;
 /// (a booting node only self-forms when NO seed answers InitJoin), so a peer booting after this
 /// node self-formed simply joins it.</para>
 ///
-/// <para><b>Races are benign.</b> If the join completes between window expiry and
-/// <c>Cluster.Join(SelfAddress)</c>, Akka ignores the join — a node joins a cluster at most once
-/// per incarnation. The residual risk is both pair nodes cold-starting inside the window while
-/// mutually unreachable (a boot-time partition): both self-form, the same dual-active class the
-/// auto-down downing strategy already accepts, with the same recovery (restart one side).</para>
+/// <para><b>Reachability guard (added 2026-07-22 after a live-gate failure).</b> An expired window
+/// is NOT sufficient evidence that the peer is down, and joining on the timer alone islands nodes.
+/// Drilled on the docker-dev rig: a node bounced by a manual failover restarted, received
+/// <c>InitJoinAck</c> from its live peer — a join in flight and healthy — but did not get the
+/// Welcome inside the window, because the peer's ring still held the node's previous incarnation
+/// (Exiting → Down → Removed). The fallback fired anyway and the node formed a SECOND cluster,
+/// islanded until an operator restarted it. So <c>Join(SelfAddress)</c> is <b>not</b> ignored
+/// mid-handshake — it wins, which is the opposite of what the original design assumed. Before
+/// self-forming, this therefore TCP-probes the other seed addresses: if any accepts a connection
+/// the peer is alive, a join is likely already in flight, and the fallback waits another window
+/// instead. That is also exactly what the fallback claims to detect — "no seed answered InitJoin
+/// (peer down at boot)".</para>
+///
+/// <para><b>Residual risk.</b> Both pair nodes cold-starting inside the window while mutually
+/// unreachable (a boot-time partition): both self-form, the same dual-active class the auto-down
+/// downing strategy already accepts, with the same recovery (restart one side).</para>
 ///
 /// <para>Deliberately duplicated from ScadaBridge's equivalent (like the termination watchdogs) —
 /// the two repos share the behavior, not a package.</para>
@@ -39,7 +50,16 @@ public static class ClusterBootstrapFallback
     /// <param name="system">The actor system whose cluster membership is being watched.</param>
     /// <param name="options">The bound cluster options carrying the window and the seed list.</param>
     /// <param name="logger">Logger for the armed / inert / self-forming decisions.</param>
-    public static void Arm(ActorSystem system, AkkaClusterOptions options, ILogger logger)
+    /// <param name="peerProbe">
+    /// Overrides the reachability probe (host, port) → reachable. Production passes
+    /// <see langword="null"/> for the real TCP connect; tests substitute it to drive the guard
+    /// deterministically.
+    /// </param>
+    public static void Arm(
+        ActorSystem system,
+        AkkaClusterOptions options,
+        ILogger logger,
+        Func<string, int, Task<bool>>? peerProbe = null)
     {
         ArgumentNullException.ThrowIfNull(system);
         ArgumentNullException.ThrowIfNull(options);
@@ -66,28 +86,108 @@ public static class ClusterBootstrapFallback
             return;
         }
 
+        // Every seed that is not this node. These are the addresses whose reachability decides
+        // whether an expired window means "peer is down" or "peer is alive and we are mid-join".
+        var peerSeeds = options.SeedNodes
+            .Select(s => TryParseAddress(s, out var a) ? a : null)
+            .Where(a => a is not null && !a.Equals(self))
+            .Select(a => (Host: a!.Host ?? string.Empty, Port: a!.Port ?? 0))
+            .Where(p => !string.IsNullOrWhiteSpace(p.Host) && p.Port > 0)
+            .ToList();
+
+        var probe = peerProbe ?? TryConnectAsync;
+
         var joined = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         cluster.RegisterOnMemberUp(() => joined.TrySetResult());
 
         _ = Task.Run(async () =>
         {
-            var winner = await Task.WhenAny(joined.Task, Task.Delay(window));
-            if (winner == joined.Task || system.WhenTerminated.IsCompleted)
+            while (true)
             {
+                var winner = await Task.WhenAny(joined.Task, Task.Delay(window));
+                if (winner == joined.Task || system.WhenTerminated.IsCompleted)
+                {
+                    return;
+                }
+
+                var reachable = await AnyReachableAsync(peerSeeds, probe);
+                if (reachable is not null)
+                {
+                    // Do NOT self-form. A reachable peer means either a join already in flight (the
+                    // failure this guard exists for) or a peer about to answer — self-forming here
+                    // creates a second cluster that can never merge.
+                    logger.LogInformation(
+                        "Self-form window ({Window}) expired, but seed peer {Peer} is reachable — a join is "
+                        + "most likely already in flight (a node re-joining after a restart is not admitted "
+                        + "until its previous incarnation is removed). Continuing to wait rather than "
+                        + "forming a second cluster.",
+                        window,
+                        reachable);
+                    continue;
+                }
+
+                logger.LogWarning(
+                    "No cluster membership after {Window} and no seed peer is reachable — the peer is down "
+                    + "at boot. Self-forming a cluster at {Self} so this node becomes operational; if the "
+                    + "peer was merely partitioned (not dead), the pair is now dual-active — restart one "
+                    + "side after the partition heals (accepted availability-first trade, decision "
+                    + "2026-07-22).",
+                    window,
+                    self);
+                cluster.Join(self);
                 return;
             }
-
-            logger.LogWarning(
-                "No cluster membership after {Window} — no seed answered InitJoin (peer down at boot). "
-                + "Self-forming a cluster at {Self} so this node becomes operational; if the peer was "
-                + "merely partitioned (not dead), the pair is now dual-active — restart one side after "
-                + "the partition heals (accepted availability-first trade, decision 2026-07-22).",
-                window,
-                self);
-            cluster.Join(self);
         });
     }
 
+    /// <summary>How long a single peer-reachability connect attempt may take.</summary>
+    public static readonly TimeSpan ProbeTimeout = TimeSpan.FromSeconds(2);
+
+    /// <summary>
+    /// Returns the first reachable peer as <c>host:port</c>, or <see langword="null"/> when none
+    /// answered — including when there are no peer seeds at all (a lone-seed config, where an
+    /// expired window really does mean this node is on its own).
+    /// </summary>
+    private static async Task<string?> AnyReachableAsync(
+        IReadOnlyList<(string Host, int Port)> peers,
+        Func<string, int, Task<bool>> probe)
+    {
+        foreach (var (host, port) in peers)
+        {
+            bool ok;
+            try
+            {
+                ok = await probe(host, port);
+            }
+            catch (Exception)
+            {
+                // A probe that cannot even be attempted (DNS gone, socket exhaustion) is treated as
+                // unreachable — it must not throw out of the watchdog and disarm the fallback.
+                ok = false;
+            }
+
+            if (ok) return $"{host}:{port}";
+        }
+
+        return null;
+    }
+
+    private static async Task<bool> TryConnectAsync(string host, int port)
+    {
+        using var client = new System.Net.Sockets.TcpClient();
+        using var cts = new CancellationTokenSource(ProbeTimeout);
+        try
+        {
+            await client.ConnectAsync(host, port, cts.Token);
+            return client.Connected;
+        }
+        catch (Exception)
+        {
+            // Refused, unresolvable, or timed out — all mean "not reachable right now".
+            return false;
+        }
+    }
+
     private static bool TryParseAddress(string seed, out Address address)
     {
         try