test(validation): ScadaBridge consumer matrix — failover suite, UNS subscribe, ConditionRefresh after reconnect (ScadaBridge#125)

Probe 'failover <primary> <backup> <mode> <dir>' drives ScadaBridge's own DataConnectionActor
(production primary/backup machine, real DataConnectionFactory, default DataConnectionOptions,
FailoverRetryCount 3) against two servers; run.py loses the primary (SIGKILL or SIGSTOP) once the
probe reports a steady primary, and the probe requires Bad quality on loss, "Connected to backup",
the backup's own value, a fresh ConditionRefresh replaying the native condition, and a write via the
backup. rig-failover.sh runs the same probe against the local docker-dev central pair.

Matrix legs gain a UNS subscription case and a Server-object alarm subscription on the reconnected
session whose ConditionRefresh must replay the active condition (11 cases). run.py gains --versions,
--suites, --failover-modes and --failover-loss.

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

scripts/validation/scadabridge-sdk-compatibility/Failover.cs

@@ -0,0 +1,225 @@
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.Reflection;
+using System.Text.Json;
+using Akka.Actor;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
+using ZB.MOM.WW.ScadaBridge.Commons.Serialization;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
+
+/// <summary>
+/// ScadaBridge#125 redundancy leg. Drives ScadaBridge's own <see cref="DataConnectionActor"/> — the
+/// production primary/backup state machine, the real <see cref="DataConnectionFactory"/>, its production
+/// <see cref="DataConnectionOptions"/> defaults and <c>FailoverRetryCount = 3</c> — against two servers.
+/// Phase 1 proves a steady primary (tag values, write, Server-object alarm subscription + ConditionRefresh).
+/// The probe then writes <c>primary-ready</c> and waits for <c>primary-killed</c>; whoever runs it (run.py,
+/// or a person against the docker-dev pair) kills the primary abruptly in between. Phase 2 requires: Bad
+/// quality pushed on loss, the actor reporting "Connected to backup", Good values from the backup, a fresh
+/// ConditionRefresh on the backup session, and (optionally) a write through the backup.
+/// </summary>
+internal static class Failover
+{
+    const string Connection = "sdk2-pair";
+
+    static string Env(string name, string fallback) =>
+        Environment.GetEnvironmentVariable(name) is { Length: > 0 } v ? v : fallback;
+
+    static void Require(bool value, string reason) { if (!value) throw new InvalidOperationException(reason); }
+
+    static bool Is(object? value, long expected)
+    {
+        try { return value is not null and not string && Convert.ToInt64(value) == expected; }
+        catch (Exception) { return false; }
+    }
+
+    public static async Task<int> RunAsync(string primary, string backup, string root, OpcUaGlobalOptions options,
+        Func<string, OpcUaEndpointConfig> configFor)
+    {
+        // Defaults fit the opcua-sdk-compatibility ProcessHost fixture; the docker-dev rig overrides them.
+        var tags = Env("SCADABRIDGE_PROBE_FAILOVER_TAGS",
+            "nsu=https://zb.com/otopcua/raw;s=Matrix/dev/Scalar,nsu=https://zb.com/otopcua/uns;s=Eq/Scalar").Split(',');
+        var writeTag = Env("SCADABRIDGE_PROBE_FAILOVER_WRITE_TAG", tags[0]);              // "none" = no writes
+        var backupValue = Env("SCADABRIDGE_PROBE_FAILOVER_BACKUP_VALUE", "17");            // "none" = not checked
+        var alarmMessage = Env("SCADABRIDGE_PROBE_FAILOVER_ALARM_PREFIX", "matrix-");     // "none" = refresh only
+        var killWait = TimeSpan.FromSeconds(int.Parse(Env("SCADABRIDGE_PROBE_FAILOVER_KILL_WAIT_S", "90")));
+        var failoverWait = TimeSpan.FromSeconds(int.Parse(Env("SCADABRIDGE_PROBE_FAILOVER_WAIT_S", "240")));
+
+        var cases = new List<object>();
+        var failures = 0;
+        var events = new ConcurrentQueue<(DateTimeOffset At, object Message)>();
+        var health = DispatchProxy.Create<ISiteHealthCollector, HealthRecorder>();
+        var calls = ((HealthRecorder)(object)health).Calls;
+        var dcOptions = new DataConnectionOptions(); // production defaults (ReconnectInterval 5 s, ...)
+        var factory = new DataConnectionFactory(NullLoggerFactory.Instance, Options.Create(options), null, Options.Create(dcOptions));
+        var primaryConfig = OpcUaEndpointConfigSerializer.ToFlatDict(configFor(primary));
+        var backupConfig = OpcUaEndpointConfigSerializer.ToFlatDict(configFor(backup));
+        DateTimeOffset? killedAt = null, backupConnectedAt = null;
+        var timings = new Dictionary<string, double>();
+
+        async Task Check(string name, Func<Task> action)
+        {
+            try { await action(); cases.Add(new { name, status = "passed" }); }
+            catch (Exception e) { failures++; cases.Add(new { name, status = "failed", error = e.ToString() }); }
+        }
+        async Task<DateTimeOffset> Until(Func<DateTimeOffset?> found, TimeSpan limit, string what)
+        {
+            var stop = DateTime.UtcNow + limit;
+            while (DateTime.UtcNow < stop)
+            {
+                if (found() is { } at) return at;
+                await Task.Delay(50);
+            }
+            throw new TimeoutException($"{what}: not observed within {limit.TotalSeconds:F0} s");
+        }
+        DateTimeOffset? Event<T>(DateTimeOffset after, Func<T, bool> match) =>
+            events.Where(e => e.At >= after && e.Message is T t && match(t)).Select(e => (DateTimeOffset?)e.At).FirstOrDefault();
+        DateTimeOffset? Health(DateTimeOffset after, string label) =>
+            calls.Where(c => c.At >= after && c.Method == nameof(ISiteHealthCollector.UpdateConnectionEndpoint) && Equals(c.Args.ElementAtOrDefault(1), label))
+                .Select(c => (DateTimeOffset?)c.At).FirstOrDefault();
+        bool GoodValue(TagValueUpdate u, string tag) => u.TagPath == tag && u.Quality == QualityCode.Good;
+
+        using var system = ActorSystem.Create("probe", "akka.loglevel = INFO\nakka.stdout-loglevel = INFO");
+        var instance = system.ActorOf(Props.Create(() => new Collector(events)), "instance");
+        var adapter = factory.Create("OpcUa", primaryConfig);
+        var actor = system.ActorOf(Props.Create(() => new DataConnectionActor(
+            Connection, adapter, dcOptions, health, factory, "OpcUa", primaryConfig, backupConfig, 3, null)), Connection);
+        var started = DateTimeOffset.UtcNow;
+        actor.Tell(new SubscribeTagsRequest("tags", "probe-instance", Connection, tags, started), instance);
+        actor.Tell(new SubscribeAlarmsRequest("alarms", "probe-instance", Connection, "i=2253", null, started), instance);
+
+        await Check("primary-connected", async () => await Until(() => Health(started, "Connected to primary"), TimeSpan.FromSeconds(60), "Connected to primary"));
+        await Check("primary-subscriptions-accepted", async () =>
+        {
+            await Until(() => Event<SubscribeTagsResponse>(started, r => r.Success), TimeSpan.FromSeconds(30), "SubscribeTagsResponse success");
+            await Until(() => Event<SubscribeAlarmsResponse>(started, r => r.Success), TimeSpan.FromSeconds(30), "SubscribeAlarmsResponse success");
+        });
+        await Check("primary-good-values-all-tags", async () =>
+        {
+            foreach (var tag in tags) await Until(() => Event<TagValueUpdate>(started, u => GoodValue(u, tag)), TimeSpan.FromSeconds(30), "Good value for " + tag);
+        });
+        await Check("primary-alarm-condition-refresh", async () =>
+        {
+            await Until(() => Event<NativeAlarmTransitionUpdate>(started, a => a.Transition.Kind == AlarmTransitionKind.SnapshotComplete), TimeSpan.FromSeconds(30), "SnapshotComplete on primary");
+            if (alarmMessage != "none")
+                await Until(() => Event<NativeAlarmTransitionUpdate>(started, a => a.Transition.Message.StartsWith(alarmMessage, StringComparison.Ordinal)), TimeSpan.FromSeconds(30), "native alarm on primary");
+        });
+        if (writeTag != "none")
+            await Check("primary-write", async () =>
+            {
+                var at = DateTimeOffset.UtcNow;
+                var response = await actor.Ask<WriteTagResponse>(new WriteTagRequest("w1", Connection, writeTag, 412, at), TimeSpan.FromSeconds(30));
+                Require(response.Success, response.ErrorMessage ?? "write refused");
+                await Until(() => Event<TagValueUpdate>(at, u => GoodValue(u, writeTag) && Is(u.Value, 412)), TimeSpan.FromSeconds(30), "412 echoed by the primary");
+            });
+
+        // Loss evidence is matched from the moment the probe hands the primary over to be killed, not from
+        // the recorded kill time: the consumer can see the socket close before the killer timestamps it.
+        var readyAt = DateTimeOffset.UtcNow;
+        if (failures == 0)
+        {
+            File.WriteAllText(Path.Combine(root, "primary-ready"), readyAt.ToString("O"));
+            Console.WriteLine("PRIMARY-READY " + Path.Combine(root, "primary-ready"));
+            var killedFile = Path.Combine(root, "primary-killed");
+            await Check("primary-killed-by-runner", async () =>
+            {
+                // The marker carries the kill time; a half-written marker is simply read again.
+                killedAt = await Until(() => File.Exists(killedFile) && DateTimeOffset.TryParse(File.ReadAllText(killedFile).Trim(), out var at) ? at : null,
+                    killWait, "primary-killed marker");
+            });
+        }
+        if (killedAt is { } k)
+        {
+            await Check("loss-pushes-bad-quality", async () =>
+            {
+                var at = await Until(() => Event<TagValueUpdate>(readyAt, u => u.Quality != QualityCode.Good)
+                    ?? Event<ConnectionQualityChanged>(readyAt, q => q.Quality != QualityCode.Good), failoverWait, "Bad quality after primary loss");
+                timings["killToBadQualitySeconds"] = (at - k).TotalSeconds;
+            });
+            await Check("failover-connected-to-backup", async () =>
+            {
+                backupConnectedAt = await Until(() => Health(readyAt, "Connected to backup"), failoverWait, "Connected to backup");
+                timings["killToBackupConnectedSeconds"] = (backupConnectedAt.Value - k).TotalSeconds;
+            });
+        }
+        if (backupConnectedAt is { } b)
+        {
+            await Check("backup-good-values-all-tags", async () =>
+            {
+                foreach (var tag in tags)
+                {
+                    var at = await Until(() => Event<TagValueUpdate>(b, u => GoodValue(u, tag)), TimeSpan.FromSeconds(60), "Good value from backup for " + tag);
+                    timings["killToGood:" + tag] = (at - killedAt!.Value).TotalSeconds;
+                }
+                // The fixture's backup never saw the primary's 412, so its own value proves the feed's source.
+                if (backupValue != "none" && writeTag != "none")
+                {
+                    var first = events.Where(e => e.At >= b && e.Message is TagValueUpdate u && GoodValue(u, writeTag)).Select(e => (TagValueUpdate)e.Message).First();
+                    Require(Is(first.Value, long.Parse(backupValue)), $"First Good value after failover was {first.Value}, expected the backup's own {backupValue}");
+                }
+            });
+            await Check("backup-alarm-condition-refresh", async () =>
+            {
+                var at = await Until(() => Event<NativeAlarmTransitionUpdate>(b, a => a.Transition.Kind == AlarmTransitionKind.SnapshotComplete), TimeSpan.FromSeconds(60), "SnapshotComplete on backup");
+                timings["killToBackupSnapshotCompleteSeconds"] = (at - killedAt!.Value).TotalSeconds;
+                if (alarmMessage != "none")
+                    await Until(() => Event<NativeAlarmTransitionUpdate>(b, a => a.Transition.Kind == AlarmTransitionKind.Snapshot
+                        && a.Transition.Message.StartsWith(alarmMessage, StringComparison.Ordinal)), TimeSpan.FromSeconds(30), "native condition replayed by the backup's ConditionRefresh");
+            });
+            if (writeTag != "none")
+                await Check("backup-write", async () =>
+                {
+                    var at = DateTimeOffset.UtcNow;
+                    var response = await actor.Ask<WriteTagResponse>(new WriteTagRequest("w2", Connection, writeTag, 913, at), TimeSpan.FromSeconds(30));
+                    Require(response.Success, response.ErrorMessage ?? "write refused");
+                    await Until(() => Event<TagValueUpdate>(at, u => GoodValue(u, writeTag) && Is(u.Value, 913)), TimeSpan.FromSeconds(30), "913 echoed by the backup");
+                });
+        }
+
+        await system.Terminate();
+        var sdkAssembly = typeof(Opc.Ua.NodeId).Assembly;
+        var result = new
+        {
+            leg = "failover",
+            consumer = typeof(DataConnectionActor).Assembly.GetName().Version?.ToString(),
+            sdk = FileVersionInfo.GetVersionInfo(sdkAssembly.Location).ProductVersion,
+            process = Environment.ProcessId, primary, backup, autoAccept = false,
+            failoverRetryCount = 3, reconnectIntervalSeconds = dcOptions.ReconnectInterval.TotalSeconds,
+            killedAt, backupConnectedAt, timings, cases, failures,
+            endpointLabels = calls.Where(c => c.Method == nameof(ISiteHealthCollector.UpdateConnectionEndpoint))
+                .Select(c => new { at = c.At, label = c.Args.ElementAtOrDefault(1)?.ToString() }),
+            alarmTransitions = events.Where(e => e.Message is NativeAlarmTransitionUpdate)
+                .Select(e => new { at = e.At, kind = ((NativeAlarmTransitionUpdate)e.Message).Transition.Kind.ToString(), message = ((NativeAlarmTransitionUpdate)e.Message).Transition.Message, source = ((NativeAlarmTransitionUpdate)e.Message).Transition.SourceReference }),
+            alarmSourceUnavailable = events.Where(e => e.Message is NativeAlarmSourceUnavailable).Select(e => e.At),
+            values = events.Where(e => e.Message is TagValueUpdate).Select(e => new { at = e.At, tag = ((TagValueUpdate)e.Message).TagPath, value = ((TagValueUpdate)e.Message).Value?.ToString(), quality = ((TagValueUpdate)e.Message).Quality.ToString() }),
+        };
+        File.WriteAllText(Path.Combine(root, "result.json"), JsonSerializer.Serialize(result, new JsonSerializerOptions { WriteIndented = true }));
+        Console.WriteLine($"Failover probe: {cases.Count - failures} passed, {failures} failed; {Path.Combine(root, "result.json")}");
+        return failures == 0 && killedAt is not null ? 0 : 1;
+    }
+
+    /// <summary>Stands in for the Instance Actor: records every message the connection actor sends it.</summary>
+    sealed class Collector : ReceiveActor
+    {
+        public Collector(ConcurrentQueue<(DateTimeOffset, object)> sink) => ReceiveAny(m => sink.Enqueue((DateTimeOffset.UtcNow, m)));
+    }
+}
+
+/// <summary>Records every <see cref="ISiteHealthCollector"/> call (the actor's own endpoint/health edges).</summary>
+public class HealthRecorder : DispatchProxy
+{
+    public ConcurrentQueue<(DateTimeOffset At, string Method, object?[] Args)> Calls { get; } = new();
+
+    protected override object? Invoke(MethodInfo? targetMethod, object?[]? args)
+    {
+        Calls.Enqueue((DateTimeOffset.UtcNow, targetMethod!.Name, args ?? []));
+        var type = targetMethod.ReturnType;
+        return type == typeof(void) || !type.IsValueType ? null : Activator.CreateInstance(type);
+    }
+}

scripts/validation/scadabridge-sdk-compatibility/Program.cs

@@ -34,10 +34,13 @@ if (args.Length == 3 && args[0] == "prepare")
     Console.WriteLine($"Consumer identity {publicCert.Thumbprint} {publicCert.Subject}");
     return 0;
 }
-if (args.Length != 3) throw new ArgumentException("Usage: Probe <endpoint> <None|Sign|SignAndEncrypt> <new-result-directory> | Probe prepare <new-pki-directory> <public-der-out>");
-var endpoint = args[0];
-var mode = Enum.Parse<OpcUaSecurityMode>(args[1]);
-var root = Path.GetFullPath(args[2]);
+// ScadaBridge#125: "failover <primary> <backup> <mode> <dir>" drives the real DataConnectionActor
+// primary/backup machine (Failover.cs); the runner kills the primary once the probe reports it is steady.
+var failoverMode = args.Length == 5 && args[0] == "failover";
+if (!failoverMode && args.Length != 3) throw new ArgumentException("Usage: Probe <endpoint> <None|Sign|SignAndEncrypt> <new-result-directory> | Probe failover <primary> <backup> <mode> <new-result-directory> | Probe prepare <new-pki-directory> <public-der-out>");
+var endpoint = failoverMode ? args[1] : args[0];
+var mode = Enum.Parse<OpcUaSecurityMode>(failoverMode ? args[3] : args[1]);
+var root = Path.GetFullPath(failoverMode ? args[4] : args[2]);
 if (Directory.Exists(root)) throw new IOException("Refusing to reuse an evidence/PKI directory.");
 Directory.CreateDirectory(root);
 static void Require(bool value, string reason) { if (!value) throw new InvalidOperationException(reason); }
@@ -50,25 +53,32 @@ var pki = Environment.GetEnvironmentVariable("SCADABRIDGE_PROBE_PKI") is { Lengt
 var options = IdentityOptions(pki);
 var ownBefore = Directory.Exists(Path.Combine(pki, "own", "certs")) ? Directory.GetFiles(Path.Combine(pki, "own", "certs")).Order().ToArray() : [];
 // Populate only this probe's trust list; never change the consumer's existing stores.
+// Several server certificate directories (a failover pair) are separated by Path.PathSeparator.
 var trustSource = Environment.GetEnvironmentVariable("SCADABRIDGE_PROBE_SERVER_CERTS");
 if (!string.IsNullOrEmpty(trustSource))
 {
     var destination = Path.Combine(options.TrustedPeerStorePath, "certs");
     Directory.CreateDirectory(destination);
-    foreach (var cert in Directory.GetFiles(trustSource, "*.der")) File.Copy(cert, Path.Combine(destination, Path.GetFileName(cert)));
+    foreach (var source in trustSource.Split(Path.PathSeparator, StringSplitOptions.RemoveEmptyEntries))
+        foreach (var cert in Directory.GetFiles(source, "*.der")) File.Copy(cert, Path.Combine(destination, Path.GetFileName(cert)), overwrite: true);
 }
+// The rig variant (a deployed OtOpcUa pair, OpcUa:AllowAnonymous=true) sets SCADABRIDGE_PROBE_ANONYMOUS=1.
+OpcUaEndpointConfig ConfigFor(string url) => new()
+{
+    EndpointUrl = url, SecurityMode = mode, AutoAcceptUntrustedCerts = false,
+    PublishingIntervalMs = 100, SamplingIntervalMs = 100,
+    UserIdentity = Environment.GetEnvironmentVariable("SCADABRIDGE_PROBE_ANONYMOUS") == "1"
+        ? new OpcUaUserIdentityConfig { TokenType = OpcUaUserTokenType.Anonymous }
+        : new OpcUaUserIdentityConfig
+        {
+            TokenType = OpcUaUserTokenType.UsernamePassword, Username = "apikey",
+            Password = Environment.GetEnvironmentVariable("OPCUA_MATRIX_PASSWORD") ?? throw new InvalidOperationException("Missing task credential"),
+        },
+};
+if (failoverMode) return await Failover.RunAsync(args[1], args[2], root, options, ConfigFor);
 var factory = new RealOpcUaClientFactory(options, NullLoggerFactory.Instance);
 await using var adapter = new OpcUaDataConnection(factory, NullLogger<OpcUaDataConnection>.Instance);
-var config = new OpcUaEndpointConfig
-{
-    EndpointUrl = endpoint, SecurityMode = mode, AutoAcceptUntrustedCerts = false,
-    PublishingIntervalMs = 100, SamplingIntervalMs = 100,
-    UserIdentity = new OpcUaUserIdentityConfig
-    {
-        TokenType = OpcUaUserTokenType.UsernamePassword, Username = "apikey",
-        Password = Environment.GetEnvironmentVariable("OPCUA_MATRIX_PASSWORD") ?? throw new InvalidOperationException("Missing task credential"),
-    },
-};
+var config = ConfigFor(endpoint);
 async Task Check(string name, Func<Task> action)
 {
     try { await action(); cases.Add(new { name, status = "passed" }); }
@@ -120,6 +130,16 @@ if (failures == 0)
         Require(actual.Quality.ToString() == "Good", "Quality was not Good");
         await adapter.UnsubscribeAsync(handle, ct);
     });
+    // ScadaBridge#125: subscriptions on the UNS NodeId too (Eq/Scalar is a separate UNS variable).
+    await Check("actual-adapter-monitor-uns", async () =>
+    {
+        var values = new ConcurrentQueue<TagValue>();
+        var handle = await adapter.SubscribeAsync(uns + "Scalar", (_, value) => values.Enqueue(value), ct);
+        Require((await adapter.WriteAsync(uns + "Scalar", 714, ct)).Success, "Monitored UNS write denied");
+        await Wait(() => values.Any(x => x.Value is int i && i == 714));
+        Require(values.Last(x => x.Value is int i && i == 714).Quality.ToString() == "Good", "UNS quality was not Good");
+        await adapter.UnsubscribeAsync(handle, ct);
+    });
     await Check("actual-adapter-alarm-filter-refresh", async () =>
     {
         var alarms = new ConcurrentQueue<NativeAlarmTransition>();
@@ -133,6 +153,19 @@ if (failures == 0)
         await adapter.DisconnectAsync(ct);
         await adapter.ConnectAsync(OpcUaEndpointConfigSerializer.ToFlatDict(config), ct);
         Require((await adapter.ReadAsync(raw + "Scalar", ct)).Success, "Consumer failed after reconnect");
+    });
+    // ScadaBridge#125: the Server-object alarm subscription on the NEW session must get its own
+    // ConditionRefresh — the active native condition is REPLAYED (Snapshot kind) and the refresh completes.
+    await Check("actual-adapter-alarm-refresh-after-reconnect", async () =>
+    {
+        var alarms = new ConcurrentQueue<NativeAlarmTransition>();
+        var handle = await adapter.SubscribeAlarmsAsync("i=2253", null, alarms.Enqueue, ct);
+        await Wait(() => alarms.Any(x => x.Kind == AlarmTransitionKind.SnapshotComplete));
+        var replayed = alarms.TakeWhile(x => x.Kind != AlarmTransitionKind.SnapshotComplete).ToArray();
+        Require(replayed.Any(x => x.Kind == AlarmTransitionKind.Snapshot && x.SourceReference.Contains("Alarm", StringComparison.Ordinal)),
+            "ConditionRefresh on the reconnected session did not replay the active native condition; saw: "
+            + string.Join(", ", replayed.Select(x => $"{x.Kind}:{x.SourceReference}:{x.Message}")));
+        await adapter.UnsubscribeAlarmsAsync(handle, ct);
         await adapter.DisconnectAsync(ct);
     });
 }

scripts/validation/scadabridge-sdk-compatibility/README.md

@@ -41,7 +41,10 @@ Never commit an entire output directory: it contains the fixture's private PKI.
 
 The consumed subset covers API-key activation, raw/UNS scalar and Int32[]
 roundtrips, durable hierarchical browse including the native alarm, data quality
-and timestamps, alarm filter/ConditionRefresh and disconnect/reconnect. History
+and timestamps on raw and UNS subscriptions, alarm filter/ConditionRefresh,
+disconnect/reconnect, and (ScadaBridge#125) a Server-object alarm subscription on
+the reconnected session whose ConditionRefresh must replay the active native
+condition (11 cases per trusted leg). History
 and Part 9 acknowledgment are not exposed by the inspected consumer interfaces;
 the parent Shared/CLI matrix covers those application services instead.
 
@@ -52,3 +55,44 @@ fixed in ScadaBridge `ff835447`/`36c64a6b`. Re-run 2026-11-12 with ScadaBridge
 ACL off; all trust negative controls failed closed. The runner retains the
 nonzero result and each actual failure; it does not loosen trust or change
 expected results to return success. No ScadaBridge production source or user PKI is modified.
+
+## Redundancy failover (ScadaBridge#125)
+
+`run.py` also runs a `failover` suite (`--suites matrix,failover`, both by
+default; `--versions candidate` runs one application version only). Each leg
+starts two independent servers and runs `Probe failover <primary> <backup>
+<mode> <dir>`, which drives ScadaBridge's own `DataConnectionActor` (the
+production primary/backup state machine, the real `DataConnectionFactory`,
+production `DataConnectionOptions` defaults, `FailoverRetryCount = 3`). Once the
+probe has a steady primary (Good values on a raw and a UNS tag, a write, the
+Server-object alarm subscription with its ConditionRefresh) it writes
+`primary-ready`; the runner then loses the primary and writes `primary-killed`.
+The probe must then see Bad quality pushed, "Connected to backup", Good values
+whose first value is the backup's own (proving the source), a fresh
+ConditionRefresh replaying the native condition, and a write through the backup.
+Two loss shapes run per security mode (`--failover-modes None,SignAndEncrypt`,
+`--failover-loss kill,freeze`):
+
+- `kill` — SIGKILL; the OS closes the primary's sockets, so the client sees the loss at once.
+- `freeze` — SIGSTOP; the sockets stay open and silent (hung host / dead link), so only the
+  consumer's session keep-alive can detect it.
+
+ScadaBridge does not read `ServiceLevel`/`ServerRedundancy`: its failover is
+endpoint-level (primary/backup connection details), so two independent fixture
+servers exercise exactly what it consumes. `rig-failover.sh <Probe.dll> <dir>`
+runs the same probe against the real local docker-dev pair (central-1 `:4840`,
+central-2 `:4841`), `docker kill`s central-1 once the primary is steady and
+starts it again afterwards. The rig admits anonymous sessions read-only and has
+no active native alarm, so that variant checks values and ConditionRefresh
+completion only. Local docker-dev only; never point it at a wonder box.
+
+Run 2026-11-12 (ScadaBridge#125) with ScadaBridge `94ae1860` (SDK client
+1.5.378.106), ACL off, against the SDK 1.5.378.182 baseline (`83ed880e`) and
+master `a28adba2` (SDK 2.0.1): matrix 11/11 in None, Sign and SignAndEncrypt on
+both, all 8 trust controls failed closed (0x80130000 / 0x801A0000); failover 8/8
+legs (11/11 cases each). Loss-to-backup timings are the same on both versions:
+`kill` about 20 s (three 5 s reconnect attempts, then the backup), `freeze` about
+20 s to detect and 86–130 s to reach the backup, because each reconnect attempt
+waits out a connect timeout against the silent primary. `rig-failover.sh` passed
+9/9 against the docker-dev pair (image SDK 2.0.1), central-1 killed, about 20 s
+to central-2.

scripts/validation/scadabridge-sdk-compatibility/rig-failover.sh

@@ -0,0 +1,46 @@
+#!/usr/bin/env bash
+# ScadaBridge#125 — the failover probe against the REAL docker-dev OtOpcUa pair (central-1 :4840 /
+# central-2 :4841), not the in-process fixture. ScadaBridge's DataConnectionActor connects to central-1
+# as primary and central-2 as backup; once it reports a steady primary, central-1's container is killed
+# (docker kill = abrupt loss) and the probe must land on central-2 with Good values and a fresh
+# ConditionRefresh. central-1 is started again afterwards. LOCAL docker-dev only — never a wonder box.
+#
+# Usage: rig-failover.sh <Probe.dll> <new-result-directory>
+# Anonymous (OpcUa:AllowAnonymous=true on the rig) cannot write there, so writes are off; the rig deploys
+# no active native alarm, so the alarm check is "ConditionRefresh completes on each node" only.
+set -euo pipefail
+probe=$1; out=$2
+primary_container=${RIG_PRIMARY_CONTAINER:-otopcua-dev-central-1-1}
+backup_container=${RIG_BACKUP_CONTAINER:-otopcua-dev-central-2-1}
+mkdir -p "$out"
+for c in "$primary_container" "$backup_container"; do
+  mkdir -p "$out/server-certs/$c"
+  docker cp "$c:/app/pki/own/certs/." "$out/server-certs/$c/"
+done
+export SCADABRIDGE_PROBE_ANONYMOUS=1
+export SCADABRIDGE_PROBE_SERVER_CERTS="$out/server-certs/$primary_container:$out/server-certs/$backup_container"
+export SCADABRIDGE_PROBE_FAILOVER_TAGS=${RIG_TAGS:-"nsu=https://zb.com/otopcua/raw;s=int3b-mb-01/plc/HR100,nsu=https://zb.com/otopcua/uns;s=a/l/e1/HR5"}
+export SCADABRIDGE_PROBE_FAILOVER_WRITE_TAG=none
+export SCADABRIDGE_PROBE_FAILOVER_BACKUP_VALUE=none
+export SCADABRIDGE_PROBE_FAILOVER_ALARM_PREFIX=none
+dotnet "$probe" failover opc.tcp://localhost:${RIG_PRIMARY_PORT:-4840} opc.tcp://localhost:${RIG_BACKUP_PORT:-4841} None "$out/consumer" \
+  > "$out/consumer.log" 2>&1 &
+pid=$!
+for _ in $(seq 1 1500); do
+  [[ -f "$out/consumer/primary-ready" ]] && break
+  kill -0 "$pid" 2>/dev/null || break
+  sleep 0.1
+done
+status=0
+if [[ -f "$out/consumer/primary-ready" ]]; then
+  python3 -c 'from datetime import datetime, timezone; print(datetime.now(timezone.utc).isoformat())' > "$out/consumer/primary-killed.tmp"
+  docker kill "$primary_container" > /dev/null
+  mv "$out/consumer/primary-killed.tmp" "$out/consumer/primary-killed"
+  wait "$pid" || status=$?
+  docker start "$primary_container" > /dev/null
+else
+  wait "$pid" || status=$?
+fi
+docker inspect "$primary_container" "$backup_container" --format '{{.Name}} {{.Config.Image}} {{.Image}} {{.State.Status}}' > "$out/containers.txt"
+echo "probe exit $status; result $out/consumer/result.json"
+exit "$status"

scripts/validation/scadabridge-sdk-compatibility/run.py

@@ -1,25 +1,169 @@
 #!/usr/bin/env python3
 """Run the real ScadaBridge consumer in separate processes against both application versions."""
 import argparse
+from datetime import datetime, timezone
 import hashlib
 import json
 import os
 from pathlib import Path
 import secrets
+import signal
 import subprocess
 import time
 
+SECURITY = {"None": "none", "Sign": "sign", "SignAndEncrypt": None}
+
+
+def leg_env(mode):
+    env = os.environ.copy()
+    env["OPCUA_MATRIX_PASSWORD"] = "otua_matrix_" + secrets.token_urlsafe(32)
+    for name in ("SCADABRIDGE_PROBE_PKI", "SCADABRIDGE_PROBE_SERVER_CERTS", "OPCUA_MATRIX_SERVER_SECURITY"):
+        env.pop(name, None)
+    if SECURITY[mode]:
+        env["OPCUA_MATRIX_SERVER_SECURITY"] = SECURITY[mode]
+    return env
+
+
+def prepare_identity(artifacts, env, leg, trusting_server_roots, result):
+    """Provision the consumer identity through ScadaBridge's own code; trust its PUBLIC cert on the given servers."""
+    pki = leg / "consumer-pki"
+    der = leg / "consumer-identity.der"
+    with (leg / "prepare.log").open("w") as log:
+        prepared = subprocess.run(["dotnet", str(artifacts["probe"]), "prepare", str(pki), str(der)],
+                                  env=env, stdout=log, stderr=subprocess.STDOUT, timeout=120)
+    if prepared.returncode != 0:
+        raise RuntimeError(f"Consumer identity provisioning failed; inspect {leg / 'prepare.log'}")
+    env["SCADABRIDGE_PROBE_PKI"] = str(pki)
+    for server_root in trusting_server_roots:
+        trusted = server_root / "server-pki" / "trusted" / "certs"
+        trusted.mkdir(parents=True)
+        (trusted / "scadabridge-consumer.der").write_bytes(der.read_bytes())
+    result["consumerIdentitySha256"] = hashlib.sha256(der.read_bytes()).hexdigest()
+
+
+def start_server(host, server_root, env, log_path):
+    command = ["dotnet", str(host), "server", str(server_root)]
+    log = log_path.open("w")
+    server = subprocess.Popen(command, env=env, stdout=log, stderr=subprocess.STDOUT)
+    until = time.monotonic() + 40
+    while not (server_root / "endpoint.txt").is_file():
+        if server.poll() is not None or time.monotonic() >= until:
+            raise RuntimeError(f"Server failed to become ready; inspect {log_path}")
+        time.sleep(0.1)
+    return server, command, (server_root / "endpoint.txt").read_text().strip()
+
+
+def stop_server(server, server_root):
+    if server.poll() is not None:
+        return server.returncode
+    server_root.mkdir(exist_ok=True)
+    (server_root / "stop").touch()
+    try:
+        server.wait(timeout=15)
+    except subprocess.TimeoutExpired:
+        server.terminate()
+        server.wait(timeout=10)
+    return server.returncode
+
+
+def matrix_leg(version, mode, trust, artifacts, root):
+    leg = root / (f"{version}-{mode}" if trust == "trusted" else f"{version}-{mode}-{trust}")
+    leg.mkdir()
+    server_root = leg / "server"
+    env = leg_env(mode)
+    result = {"suite": "matrix", "server": version, "mode": mode, "trust": trust,
+              "expect": "pass" if trust == "trusted" else "fail"}
+    if mode != "None":
+        prepare_identity(artifacts, env, leg, [] if trust == "untrusted-client" else [server_root], result)
+    server, result["serverCommand"], endpoint = start_server(artifacts[f"{version}_host"], server_root, env, leg / "server.log")
+    try:
+        if trust != "untrusted-server":
+            env["SCADABRIDGE_PROBE_SERVER_CERTS"] = str(server_root / "server-pki" / "own" / "certs")
+        command = ["dotnet", str(artifacts["probe"]), endpoint, mode, str(leg / "consumer")]
+        with (leg / "consumer.log").open("w") as log:
+            probe = subprocess.run(command, env=env, stdout=log, stderr=subprocess.STDOUT, timeout=150)
+        result.update(command=command, exit=probe.returncode, result=str(leg / "consumer" / "result.json"))
+    finally:
+        result["serverExit"] = stop_server(server, server_root)
+    result["verdict"] = "passed" if result["serverExit"] == 0 and (result["exit"] == 0) == (result["expect"] == "pass") else "failed"
+    return result
+
+
+def failover_leg(version, mode, loss, artifacts, root):
+    """ScadaBridge#125: two independent servers; the consumer's DataConnectionActor fails over when the primary is lost.
+
+    loss=kill   SIGKILL: the OS closes the primary's sockets, so the client sees the loss at once.
+    loss=freeze SIGSTOP: the primary's sockets stay open and silent (a hung host / dead link), so only the
+                consumer's session keep-alive can detect it.
+    """
+    leg = root / f"{version}-failover-{mode}-{loss}"
+    leg.mkdir()
+    roots = {"primary": leg / "primary", "backup": leg / "backup"}
+    env = leg_env(mode)
+    result = {"suite": "failover", "server": version, "mode": mode, "loss": loss, "trust": "trusted", "expect": "pass"}
+    if mode != "None":
+        prepare_identity(artifacts, env, leg, list(roots.values()), result)
+    servers, endpoints = {}, {}
+    try:
+        for name, server_root in roots.items():
+            servers[name], result[f"{name}Command"], endpoints[name] = start_server(
+                artifacts[f"{version}_host"], server_root, env, leg / f"{name}.log")
+        # Even a None-mode client validates the server certificate the endpoint hands it.
+        env["SCADABRIDGE_PROBE_SERVER_CERTS"] = os.pathsep.join(
+            str(r / "server-pki" / "own" / "certs") for r in roots.values())
+        consumer = leg / "consumer"
+        command = ["dotnet", str(artifacts["probe"]), "failover", endpoints["primary"], endpoints["backup"], mode, str(consumer)]
+        result["command"] = command
+        with (leg / "consumer.log").open("w") as log:
+            probe = subprocess.Popen(command, env=env, stdout=log, stderr=subprocess.STDOUT)
+            until = time.monotonic() + 150
+            while not (consumer / "primary-ready").is_file() and probe.poll() is None and time.monotonic() < until:
+                time.sleep(0.1)
+            if (consumer / "primary-ready").is_file():
+                killed_at = datetime.now(timezone.utc).isoformat()  # taken BEFORE the signal
+                if loss == "kill":
+                    servers["primary"].kill()  # abrupt loss: no graceful session close
+                    servers["primary"].wait(timeout=10)
+                else:
+                    os.kill(servers["primary"].pid, signal.SIGSTOP)
+                (consumer / "primary-killed").write_text(killed_at)
+                result["primaryKilledAt"] = killed_at
+            probe.wait(timeout=480)
+        result.update(exit=probe.returncode, result=str(consumer / "result.json"))
+    finally:
+        if loss == "freeze" and "primary" in servers and servers["primary"].poll() is None:
+            servers["primary"].kill()  # a frozen process cannot honour the stop file
+            servers["primary"].wait(timeout=10)
+        for name, server in servers.items():
+            result[f"{name}Exit"] = stop_server(server, roots[name])
+    killed = "primaryKilledAt" in result
+    result["serverExit"] = result.get("backupExit")
+    result["verdict"] = "passed" if killed and result.get("exit") == 0 and result.get("backupExit") == 0 else "failed"
+    return result
+
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--baseline-host", type=Path, required=True)
-    parser.add_argument("--candidate-host", type=Path, required=True)
+    parser.add_argument("--baseline-host", type=Path)
+    parser.add_argument("--candidate-host", type=Path)
     parser.add_argument("--probe", type=Path, required=True)
     parser.add_argument("--results", type=Path, required=True)
+    parser.add_argument("--versions", default="baseline,candidate",
+                        help="comma list of baseline/candidate (each needs its --<version>-host)")
+    parser.add_argument("--suites", default="matrix,failover", help="comma list of matrix/failover")
+    parser.add_argument("--failover-modes", default="None,SignAndEncrypt")
+    parser.add_argument("--failover-loss", default="kill,freeze", help="comma list of kill/freeze")
     args = parser.parse_args()
+    versions = [v for v in args.versions.split(",") if v]
+    suites = [s for s in args.suites.split(",") if s]
     root = args.results.resolve()
     root.mkdir(parents=True, exist_ok=False)
-    artifacts = {name: path.resolve() for name, path in vars(args).items() if name != "results"}
+    artifacts = {"probe": args.probe.resolve()}
+    for version in versions:
+        host = getattr(args, f"{version}_host")
+        if host is None:
+            raise SystemExit(f"--{version}-host is required for version {version}")
+        artifacts[f"{version}_host"] = host.resolve()
     for path in artifacts.values():
         if not path.is_file():
             raise FileNotFoundError(path)
@@ -34,69 +178,18 @@ def main():
     #   untrusted-server  - the consumer's trusted store does not hold the server certificate
     legs = [(m, "trusted") for m in ("None", "Sign", "SignAndEncrypt")]
     legs += [(m, n) for m in ("Sign", "SignAndEncrypt") for n in ("untrusted-client", "untrusted-server")]
-    for version in ("baseline", "candidate"):
-        for mode, trust in legs:
-            leg = root / (f"{version}-{mode}" if trust == "trusted" else f"{version}-{mode}-{trust}")
-            leg.mkdir()
-            server_root = leg / "server"
-            env = os.environ.copy()
-            env["OPCUA_MATRIX_PASSWORD"] = "otua_matrix_" + secrets.token_urlsafe(32)
-            env.pop("SCADABRIDGE_PROBE_PKI", None)
-            env.pop("SCADABRIDGE_PROBE_SERVER_CERTS", None)
-            if mode == "None":
-                env["OPCUA_MATRIX_SERVER_SECURITY"] = "none"
-            elif mode == "Sign":
-                env["OPCUA_MATRIX_SERVER_SECURITY"] = "sign"
-            else:
-                env.pop("OPCUA_MATRIX_SERVER_SECURITY", None)
-            server_command = ["dotnet", str(artifacts[f"{version}_host"]), "server", str(server_root)]
-            result = {"server": version, "mode": mode, "trust": trust, "expect": "pass" if trust == "trusted" else "fail",
-                      "serverCommand": server_command}
-            if mode != "None":
-                pki = leg / "consumer-pki"
-                der = leg / "consumer-identity.der"
-                with (leg / "prepare.log").open("w") as log:
-                    prepared = subprocess.run(["dotnet", str(artifacts["probe"]), "prepare", str(pki), str(der)],
-                                              env=env, stdout=log, stderr=subprocess.STDOUT, timeout=120)
-                if prepared.returncode != 0:
-                    raise RuntimeError(f"Consumer identity provisioning failed; inspect {leg / 'prepare.log'}")
-                env["SCADABRIDGE_PROBE_PKI"] = str(pki)
-                if trust != "untrusted-client":
-                    trusted = server_root / "server-pki" / "trusted" / "certs"
-                    trusted.mkdir(parents=True)
-                    (trusted / "scadabridge-consumer.der").write_bytes(der.read_bytes())
-                result["consumerIdentitySha256"] = hashlib.sha256(der.read_bytes()).hexdigest()
-            with (leg / "server.log").open("w") as server_log:
-                server = subprocess.Popen(server_command, env=env, stdout=server_log, stderr=subprocess.STDOUT)
-                try:
-                    until = time.monotonic() + 40
-                    while not (server_root / "endpoint.txt").is_file():
-                        if server.poll() is not None or time.monotonic() >= until:
-                            raise RuntimeError(f"Server failed to become ready; inspect {leg / 'server.log'}")
-                        time.sleep(0.1)
-                    endpoint = (server_root / "endpoint.txt").read_text().strip()
-                    if trust != "untrusted-server":
-                        env["SCADABRIDGE_PROBE_SERVER_CERTS"] = str(server_root / "server-pki" / "own" / "certs")
-                    command = ["dotnet", str(artifacts["probe"]), endpoint, mode, str(leg / "consumer")]
-                    with (leg / "consumer.log").open("w") as log:
-                        probe = subprocess.run(command, env=env, stdout=log, stderr=subprocess.STDOUT, timeout=150)
-                    result.update(command=command, exit=probe.returncode,
-                                  result=str(leg / "consumer" / "result.json"))
-                finally:
-                    server_root.mkdir(exist_ok=True)
-                    (server_root / "stop").touch()
-                    try:
-                        server.wait(timeout=15)
-                    except subprocess.TimeoutExpired:
-                        server.terminate()
-                        server.wait(timeout=10)
-                    result["serverExit"] = server.returncode
-            receipt["legs"].append(result)
-            (root / "receipt.json").write_text(json.dumps(receipt, indent=2))
-    # A trusted leg passes only on exit 0; a negative control passes only when the consumer fails closed.
-    for leg in receipt["legs"]:
-        leg["verdict"] = "passed" if leg["serverExit"] == 0 and (leg["exit"] == 0) == (leg["expect"] == "pass") else "failed"
-    (root / "receipt.json").write_text(json.dumps(receipt, indent=2))
+    for version in versions:
+        if "matrix" in suites:
+            for mode, trust in legs:
+                receipt["legs"].append(matrix_leg(version, mode, trust, artifacts, root))
+                (root / "receipt.json").write_text(json.dumps(receipt, indent=2))
+        if "failover" in suites:
+            for mode in args.failover_modes.split(","):
+                for loss in args.failover_loss.split(","):
+                    receipt["legs"].append(failover_leg(version, mode, loss, artifacts, root))
+                    (root / "receipt.json").write_text(json.dumps(receipt, indent=2))
+    # A trusted leg passes only on exit 0; a negative control passes only when the consumer fails closed;
+    # a failover leg passes only when the primary was really killed and the consumer passed on the backup.
     return int(any(leg["verdict"] != "passed" for leg in receipt["legs"]))