lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/ScriptedAlarmHostActorTests.cs

using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.TestKit;
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.ScriptedAlarms;

/// <summary>
/// Verifies <see cref="ScriptedAlarmHostActor"/> loads the enabled subset of
/// <see cref="EquipmentScriptedAlarmPlan"/>s into its <see cref="ScriptedAlarmEngine"/>, registers mux
/// interest for their dependency refs after the load completes, feeds live
/// <see cref="VirtualTagActor.DependencyValueChanged"/> values into the engine, and bridges the
/// engine's emissions to both an <see cref="OpcUaPublishActor.AlarmStateUpdate"/> and an
/// <see cref="AlarmTransitionEvent"/> on the cluster <c>alerts</c> topic.
/// </summary>
public sealed class ScriptedAlarmHostActorTests : RuntimeActorTestBase
{
    private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(8);

    /// <summary>Plan whose predicate compares the single tag "M.T" against 90 — enabled by default.</summary>
    private static EquipmentScriptedAlarmPlan Plan(
        string id = "alm-1",
        string equipmentId = "Plant/Line1/M",
        string name = "HighTemp",
        string depRef = "M.T",
        int threshold = 90,
        bool enabled = true,
        int severity = 800) =>
        new(
            ScriptedAlarmId: id,
            EquipmentId: equipmentId,
            Name: name,
            AlarmType: "AlarmCondition",
            Severity: severity,
            MessageTemplate: "condition",
            PredicateScriptId: $"{id}-script",
            PredicateSource: $"return (int)ctx.GetTag(\"{depRef}\").Value > {threshold};",
            DependencyRefs: new[] { depRef },
            HistorizeToAveva: true,
            Retain: true,
            Enabled: enabled);

    /// <summary>Plan whose predicate references an unknown identifier so it fails to compile — applying it
    /// faults the engine's LoadAsync. Used to prove a faulted load doesn't crash the host.</summary>
    private static EquipmentScriptedAlarmPlan BadPlan(string id = "bad-1") =>
        new(
            ScriptedAlarmId: id,
            EquipmentId: "Plant/Line1/M",
            Name: "Broken",
            AlarmType: "AlarmCondition",
            Severity: 500,
            MessageTemplate: "broken",
            PredicateScriptId: $"{id}-script",
            PredicateSource: "return unknownIdentifier;", // uncompilable → LoadAsync throws
            DependencyRefs: Array.Empty<string>(),
            HistorizeToAveva: false,
            Retain: false,
            Enabled: true);

    private static ScriptedAlarmEngine BuildEngine(DependencyMuxTagUpstreamSource upstream)
    {
        var logger = new LoggerConfiguration().CreateLogger();
        return new ScriptedAlarmEngine(upstream, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger);
    }

    /// <summary>The local node id used by the redundancy-gating tests.</summary>
    private static readonly NodeId LocalNode = new("node-A");

    private (IActorRef Host, DependencyMuxTagUpstreamSource Upstream) Spawn(
        TestProbe publish, TestProbe mux, NodeId? localNode = null)
    {
        var upstream = new DependencyMuxTagUpstreamSource();
        var engine = BuildEngine(upstream);
        var host = Sys.ActorOf(ScriptedAlarmHostActor.Props(publish.Ref, mux.Ref, upstream, engine, localNode));
        return (host, upstream);
    }

    /// <summary>Tell the host a <see cref="RedundancyStateChanged"/> snapshot marking
    /// <paramref name="nodeId"/> (defaults to <see cref="LocalNode"/>) with <paramref name="role"/>
    /// so the gate observes the local role. Pass a different <paramref name="nodeId"/> to produce a
    /// snapshot that does NOT contain the host's own node.</summary>
    private static void TellRedundancyRole(IActorRef host, RedundancyRole role, NodeId? nodeId = null)
    {
        var id = nodeId ?? LocalNode;
        host.Tell(new RedundancyStateChanged(
            new[]
            {
                new NodeRedundancyState(
                    NodeId: id,
                    Role: role,
                    IsClusterLeader: role == RedundancyRole.Primary,
                    IsDriverPrimary: role == RedundancyRole.Primary,
                    AsOfUtc: DateTime.UtcNow),
            },
            CorrelationId.NewId()));
    }

    /// <summary>Subscribe <paramref name="probe"/> to the <c>alerts</c> DPS topic and wait for the ack.
    /// The Subscribe is sent FROM the probe so the SubscribeAck returns to it.</summary>
    private void SubscribeToAlerts(TestProbe probe)
    {
        DistributedPubSub.Get(Sys).Mediator.Tell(
            new Subscribe(ScriptedAlarmHostActor.AlertsTopic, probe.Ref), probe.Ref);
        probe.ExpectMsg<SubscribeAck>(Timeout);
    }

    /// <summary>Load + interest: applying one enabled alarm registers mux interest for its dep ref
    /// AFTER the engine load completes; a disabled alarm in the same apply contributes no dep ref.</summary>
    [Fact]
    public void Apply_loads_enabled_alarm_and_registers_interest()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var (host, _) = Spawn(publish, mux);

        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[]
        {
            Plan(id: "alm-1", depRef: "M.T"),
            Plan(id: "alm-2", depRef: "M.X", enabled: false), // disabled — not loaded, dep absent
        }));

        var reg = mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);
        reg.TagRefs.ShouldContain("M.T");
        reg.TagRefs.ShouldNotContain("M.X");
    }

    /// <summary>Activation path: with the alarm loaded, pushing a value above the threshold drives an
    /// Inactive→Active transition — the host publishes an AlarmStateUpdate(Active=true, Realm: AddressSpaceRealm.Uns) and an
    /// AlarmTransitionEvent("Activated") on the alerts topic.</summary>
    [Fact]
    public void Dependency_change_above_threshold_activates_alarm()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        var now = DateTime.UtcNow;
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, now));

        var state = publish.ExpectMsg<OpcUaPublishActor.AlarmStateUpdate>(Timeout);
        state.AlarmNodeId.ShouldBe("alm-1");
        // The full Part 9 snapshot bridges through (T15) — every Core condition field maps:
        // on activation the engine sets Active, clears Ack AND Confirm (a new active occurrence needs a
        // fresh ack→clear→confirm cycle), keeps Enabled, and leaves Shelving unshelved.
        state.State.Active.ShouldBeTrue();              // Condition.Active == Active
        state.State.Acknowledged.ShouldBeFalse();       // Condition.Acked == Unacknowledged on activation
        state.State.Confirmed.ShouldBeFalse();          // Condition.Confirmed == Unconfirmed on activation
        state.State.Enabled.ShouldBeTrue();             // Condition.Enabled == Enabled
        state.State.Shelving.ShouldBe(AlarmShelvingKind.Unshelved); // Condition.Shelving.Kind == Unshelved
        state.State.Severity.ShouldBe((ushort)1000);    // 800 → Critical bucket → 1000
        state.State.Message.ShouldBe("condition");      // e.Message

        var evt = alerts.ExpectMsg<AlarmTransitionEvent>(Timeout);
        evt.AlarmId.ShouldBe("alm-1");
        evt.TransitionKind.ShouldBe("Activated");
        evt.Severity.ShouldBe(1000); // 800 → Critical bucket → 1000
        evt.User.ShouldBe("system");
        // Historian feed prep: the Part-9 subtype rides along (Plan's AlarmType "AlarmCondition"
        // parses to AlarmKind.AlarmCondition); an engine-driven Activated transition has no comment.
        evt.AlarmTypeName.ShouldBe("AlarmCondition");
        evt.Comment.ShouldBeNull();
    }

    /// <summary>Clear path: after activating, pushing a value below the threshold drives Active→Inactive
    /// — AlarmStateUpdate(Active=false, Realm: AddressSpaceRealm.Uns) + AlarmTransitionEvent("Cleared").</summary>
    [Fact]
    public void Dependency_change_below_threshold_clears_alarm()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan() }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);

        // Activate first.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
        publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        alerts.FishForMessage<AlarmTransitionEvent>(e => e.TransitionKind == "Activated", Timeout);

        // Now clear.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 10, DateTime.UtcNow));

        var cleared = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => !m.State.Active, Timeout);
        cleared.AlarmNodeId.ShouldBe("alm-1");

        var evt = alerts.FishForMessage<AlarmTransitionEvent>(e => e.TransitionKind == "Cleared", Timeout);
        evt.AlarmId.ShouldBe("alm-1");
    }

    /// <summary>Re-apply reloads: a second ApplyScriptedAlarms with a different alarm set loads the new
    /// alarm — a fresh RegisterInterest reflecting the new dependency refs lands on the mux.</summary>
    [Fact]
    public void Reapply_reloads_with_new_dependency_refs()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var (host, _) = Spawn(publish, mux);

        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T") }));
        var first = mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);
        first.TagRefs.ShouldContain("M.T");

        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-9", depRef: "M.Y") }));
        var second = mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);
        second.TagRefs.ShouldContain("M.Y");
        second.TagRefs.ShouldNotContain("M.T");
    }

    /// <summary>Faulted load resilience: applying an alarm whose predicate doesn't compile faults the
    /// engine's LoadAsync. The host logs a Warning (Status.Failure handler) and stays alive — it must
    /// still process a subsequent valid apply, registering interest for that apply's dep refs.</summary>
    [Fact]
    public void Faulted_load_does_not_crash_host()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var (host, _) = Spawn(publish, mux);

        // Apply a plan that fails to compile — LoadAsync faults, the host swallows it as a Warning and
        // does NOT register interest (no dep refs to register, and the load never completed).
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { BadPlan() }));
        mux.ExpectNoMsg(TimeSpan.FromMilliseconds(500));

        // Prove the actor is still responsive: a later valid apply loads + registers interest as normal.
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-ok", depRef: "M.T") }));
        var reg = mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);
        reg.TagRefs.ShouldContain("M.T");
    }

    /// <summary>Stale-load guard (fix A): two back-to-back applies with different dep-ref sets must end
    /// with the mux holding the SECOND (latest-generation) set — a stale earlier completion must never
    /// re-introduce the first set's refs.
    ///
    /// <para>Limitation: forcing the two LoadAsync continuations to complete out of order is not
    /// deterministic via real async timing (both loads are short + run on the thread pool). This test
    /// therefore validates the guard's observable contract rather than the race itself: it fishes for
    /// the RegisterInterest carrying the second apply's refs, then asserts no LATER RegisterInterest
    /// re-introduces the first apply's refs. With the generation guard in place the latest apply always
    /// wins; without it, an out-of-order stale completion could land a "M.A"-bearing RegisterInterest
    /// after the "M.B" one — which this assertion would catch.</para></summary>
    [Fact]
    public void Stale_load_does_not_register_superseded_dep_refs()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var (host, _) = Spawn(publish, mux);

        // Fire two applies in quick succession with disjoint dep refs; the second supersedes the first.
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-a", depRef: "M.A") }));
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-b", depRef: "M.B") }));

        // The latest generation must win: fish for the RegisterInterest reflecting the second apply.
        mux.FishForMessage<DependencyMuxActor.RegisterInterest>(r => r.TagRefs.Contains("M.B"), Timeout);

        // No LATER RegisterInterest may re-introduce the first (superseded) apply's "M.A" ref.
        mux.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    /// <summary>Command path: an AlarmCommand("Acknowledge") for an alarm this host owns (and that is
    /// currently active+unacknowledged) drives the engine's AcknowledgeAsync — observed via the resulting
    /// AlarmStateUpdate(Acknowledged=true, Realm: AddressSpaceRealm.Uns) and an AlarmTransitionEvent("Acknowledged") on the alerts topic
    /// carrying the command's User (the user threads through AcknowledgeAsync → LastAckUser → evt.User).</summary>
    [Fact]
    public void AlarmCommand_acknowledge_drives_engine_with_mapped_args()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T") }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Activate so there is something to acknowledge (Acknowledge no-ops on an already-acked alarm).
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
        publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active && !m.State.Acknowledged, Timeout);
        alerts.FishForMessage<AlarmTransitionEvent>(e => e.TransitionKind == "Activated", Timeout);

        // Acknowledge via the command topic — the host owns alm-1, so AcknowledgeAsync runs.
        host.Tell(new AlarmCommand(
            AlarmId: "alm-1", Operation: "Acknowledge", User: "alice", Comment: "ack-note", UnshelveAtUtc: null));

        var acked = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Acknowledged, Timeout);
        acked.AlarmNodeId.ShouldBe("alm-1");
        acked.State.Acknowledged.ShouldBeTrue();

        // The transition carries the acting user mapped from cmd.User (proves AcknowledgeAsync got it).
        var evt = alerts.FishForMessage<AlarmTransitionEvent>(e => e.TransitionKind == "Acknowledged", Timeout);
        evt.AlarmId.ShouldBe("alm-1");
        evt.User.ShouldBe("alice");
        // Historian feed prep: the operator's ack comment threads through to the transition event.
        evt.Comment.ShouldBe("ack-note");
        evt.AlarmTypeName.ShouldBe("AlarmCondition");
    }

    /// <summary>Ownership filter: an AlarmCommand for an AlarmId this host's engine does NOT own is ignored
    /// — the engine op never runs, so no AlarmStateUpdate and no alerts transition are produced.</summary>
    [Fact]
    public void AlarmCommand_for_unowned_alarm_is_ignored()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T") }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed; host owns only alm-1

        // Command targets an alarm this engine never loaded — it must be a no-op.
        host.Tell(new AlarmCommand(
            AlarmId: "not-mine", Operation: "Acknowledge", User: "alice", Comment: null, UnshelveAtUtc: null));

        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); // no engine op → no projection
        alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); // no engine op → no transition
    }

    /// <summary>Validation: an AddComment command with empty or whitespace comment text is rejected (logged),
    /// NOT propagated to the engine — the actor stays alive and still processes a subsequent valid command,
    /// proving it didn't fault and the engine's AddCommentAsync was never driven.</summary>
    [Theory]
    [InlineData("")]
    [InlineData("   ")]
    public void AlarmCommand_add_comment_empty_text_is_rejected_not_driven(string emptyComment)
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T") }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // AddComment with empty/whitespace text is rejected before reaching the engine.
        host.Tell(new AlarmCommand(
            AlarmId: "alm-1", Operation: "AddComment", User: "alice", Comment: emptyComment, UnshelveAtUtc: null));
        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); // rejected → no engine op → no OPC UA projection
        alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(200));  // rejected → no alerts event

        // Prove the actor survived: activate the alarm and observe the normal projection flow.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
        var state = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        state.AlarmNodeId.ShouldBe("alm-1");
    }

    /// <summary>Positive AddComment path: a non-empty AddComment for a loaded alarm drives the engine's
    /// AddCommentAsync — observed via an AlarmTransitionEvent("CommentAdded") on the alerts topic carrying
    /// the acting user (proves the op ran end-to-end through the host dispatch).</summary>
    [Fact]
    public void AlarmCommand_add_comment_nonempty_drives_engine()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T") }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // AddComment with a non-empty comment drives the engine — CommentAdded transition emitted.
        host.Tell(new AlarmCommand(
            AlarmId: "alm-1", Operation: "AddComment", User: "bob", Comment: "note from operator", UnshelveAtUtc: null));

        var evt = alerts.FishForMessage<AlarmTransitionEvent>(e => e.TransitionKind == "CommentAdded", Timeout);
        evt.AlarmId.ShouldBe("alm-1");
    }

    /// <summary>CommentAdded transition carries the commenter's identity: a non-empty AddComment drives
    /// AddCommentAsync — the resulting AlarmTransitionEvent("CommentAdded") on the alerts topic must carry
    /// <c>User == cmd.User</c> (the acting operator), NOT the generic <c>"system"</c> sentinel that
    /// engine-driven transitions use. This verifies the <see cref="ScriptedAlarmHostActor"/>
    /// TransitionUser arm for <see cref="EmissionKind.CommentAdded"/>.</summary>
    [Fact]
    public void CommentAdded_transition_carries_operator_user()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T") }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Send AddComment — the alarm does NOT need to be active; AddCommentAsync is unconditional.
        host.Tell(new AlarmCommand(
            AlarmId: "alm-1", Operation: "AddComment", User: "carol", Comment: "operator note", UnshelveAtUtc: null));

        // The CommentAdded AlarmTransitionEvent must carry the operator's identity, not "system".
        var evt = alerts.FishForMessage<AlarmTransitionEvent>(e => e.TransitionKind == "CommentAdded", Timeout);
        evt.AlarmId.ShouldBe("alm-1");
        evt.User.ShouldBe("carol");           // operator — NOT "system"
        evt.Comment.ShouldBe("operator note");
    }

    /// <summary>Validation: a TimedShelve command missing UnshelveAtUtc is rejected (logged), NOT thrown —
    /// the actor stays alive and still processes a subsequent valid command, proving it didn't fault.</summary>
    [Fact]
    public void AlarmCommand_timed_shelve_missing_unshelve_time_is_rejected_not_thrown()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T") }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // TimedShelve with a null UnshelveAtUtc is malformed — the host rejects + logs, does not throw.
        host.Tell(new AlarmCommand(
            AlarmId: "alm-1", Operation: "TimedShelve", User: "alice", Comment: null, UnshelveAtUtc: null));
        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); // rejected → no engine op → no projection

        // Prove the actor survived: activate the alarm and observe the normal projection flow.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
        var state = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        state.AlarmNodeId.ShouldBe("alm-1");
    }

    /// <summary>Default-emit (T1): before ANY RedundancyStateChanged snapshot arrives — the boot window,
    /// and the steady state for single-node deploys (the sole node is always Primary) — the host MUST
    /// publish the cluster-wide alerts transition. Constructed WITH a localNode but no snapshot sent, so
    /// the cached local role is unknown ⇒ treated as Primary/emit.</summary>
    [Fact]
    public void Emission_is_published_to_alerts_by_default_before_any_redundancy_state()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux, LocalNode);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // No RedundancyStateChanged sent — local role unknown ⇒ default-emit.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));

        // The OPC UA node write happens AND the alerts transition is published.
        publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        var evt = alerts.ExpectMsg<AlarmTransitionEvent>(Timeout);
        evt.AlarmId.ShouldBe("alm-1");
        evt.TransitionKind.ShouldBe("Activated");
    }

    /// <summary>Secondary suppression (T1): when the cached local role is Secondary, the host MUST NOT
    /// publish the cluster-wide alerts transition (the Primary publishes the single copy) — but it MUST
    /// still write the local OPC UA condition node so the secondary's address space stays warm for failover.</summary>
    [Fact]
    public void Secondary_node_suppresses_alerts_publish_but_still_writes_opcua()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux, LocalNode);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Mark this node Secondary, then activate.
        TellRedundancyRole(host, RedundancyRole.Secondary);
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));

        // The local OPC UA node write is UNGATED — it must still arrive.
        var state = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        state.AlarmNodeId.ShouldBe("alm-1");

        // The cluster-wide alerts publish is gated off on the secondary.
        alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    /// <summary>Primary publishes (T1): when the cached local role is Primary, the host publishes the
    /// cluster-wide alerts transition as normal (this is the single copy the fleet sees).</summary>
    [Fact]
    public void Primary_node_publishes_alerts()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux, LocalNode);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Mark this node Primary, then activate.
        TellRedundancyRole(host, RedundancyRole.Primary);
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));

        publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        var evt = alerts.ExpectMsg<AlarmTransitionEvent>(Timeout);
        evt.AlarmId.ShouldBe("alm-1");
        evt.TransitionKind.ShouldBe("Activated");
    }

    /// <summary>Boot-window multi-driver suppression (archreview 03/S4): role unknown + a real driver peer
    /// (member count 2) ⇒ the alerts publish is DENIED (default-deny closes the dual-primary window that would
    /// historize duplicate AVEVA rows), while the OPC UA node write stays ungated.</summary>
    [Fact]
    public void Unknown_role_multi_driver_suppresses_alerts_but_still_writes_opcua()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = SpawnWithMemberCount(publish, mux, LocalNode, driverMemberCount: 2);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);

        // No snapshot sent ⇒ role unknown; count 2 ⇒ default-DENY.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));

        publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    /// <summary>Boot-window single-driver emit (archreview 03/S4): role unknown + no driver peer (member
    /// count 1) ⇒ the alerts publish proceeds (single-node posture preserved).</summary>
    [Fact]
    public void Unknown_role_single_driver_publishes_alerts()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = SpawnWithMemberCount(publish, mux, LocalNode, driverMemberCount: 1);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);

        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));

        publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        alerts.ExpectMsg<AlarmTransitionEvent>(Timeout).AlarmId.ShouldBe("alm-1");
    }

    private (IActorRef Host, DependencyMuxTagUpstreamSource Upstream) SpawnWithMemberCount(
        TestProbe publish, TestProbe mux, NodeId localNode, int driverMemberCount)
    {
        var upstream = new DependencyMuxTagUpstreamSource();
        var engine = BuildEngine(upstream);
        var host = Sys.ActorOf(ScriptedAlarmHostActor.Props(
            publish.Ref, mux.Ref, upstream, engine, localNode, driverMemberCountProvider: () => driverMemberCount));
        return (host, upstream);
    }

    /// <summary>Inbound command ungated by role (T1): the alerts-publish gate must NOT affect inbound
    /// command processing. Under a Secondary role, an AlarmCommand("Acknowledge") for an owned, active
    /// alarm still drives the engine — observed via the resulting AlarmStateUpdate(Acknowledged=true, Realm: AddressSpaceRealm.Uns)
    /// (the OPC UA node write is ungated so the secondary's engine state + address space stay consistent).</summary>
    [Fact]
    public void Inbound_AlarmCommand_is_processed_regardless_of_role()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var (host, _) = Spawn(publish, mux, LocalNode);

        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T") }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Mark this node Secondary — the alerts publish is gated, but command processing is NOT.
        TellRedundancyRole(host, RedundancyRole.Secondary);

        // Activate so there is something to acknowledge.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
        publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active && !m.State.Acknowledged, Timeout);

        // Acknowledge via the command topic — the engine must process it even on the secondary.
        host.Tell(new AlarmCommand(
            AlarmId: "alm-1", Operation: "Acknowledge", User: "alice", Comment: "ack-note", UnshelveAtUtc: null));

        var acked = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Acknowledged, Timeout);
        acked.AlarmNodeId.ShouldBe("alm-1");
        acked.State.Acknowledged.ShouldBeTrue();
    }

    /// <summary>Detached suppression (A1): when the cached local role is Detached the host MUST NOT
    /// publish the cluster-wide alerts transition (identical gating as Secondary) — but it MUST still
    /// write the local OPC UA condition node so the detached node's address space stays warm.</summary>
    [Fact]
    public void Detached_node_suppresses_alerts_publish_but_still_writes_opcua()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux, LocalNode);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Mark this node Detached, then activate.
        TellRedundancyRole(host, RedundancyRole.Detached);
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));

        // The local OPC UA node write is UNGATED — it must still arrive.
        var state = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        state.AlarmNodeId.ShouldBe("alm-1");

        // The cluster-wide alerts publish is gated off on a detached node.
        alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    /// <summary>HistorizeToAveva flag threading (middle-link): the host MUST carry the plan's
    /// <c>HistorizeToAveva</c> flag onto the <see cref="AlarmTransitionEvent"/> it publishes to the
    /// <c>alerts</c> topic. Verifies both the <c>true</c> (default <see cref="Plan"/>) and the
    /// <c>false</c> (<see cref="BadPlan"/> carries false on the fixture) cases so any regression in
    /// <c>ScriptedAlarmHostActor.OnEngineEmission</c>'s flag threading is caught here before
    /// <c>HistorianAdapterActor</c>'s opt-out gate becomes the first line of defence.</summary>
    [Fact]
    public void HistorizeToAveva_flag_is_threaded_onto_published_AlarmTransitionEvent()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        // Build two plans: one with HistorizeToAveva=true (Plan default), one with false.
        // Use distinct ids + dep refs so both load cleanly side-by-side.
        var planTrue = Plan(id: "alm-hist-true", depRef: "H.T", severity: 800);       // HistorizeToAveva: true
        var planFalse = Plan(id: "alm-hist-false", depRef: "H.F", severity: 800) with { HistorizeToAveva = false };

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { planTrue, planFalse }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Activate the true-flag alarm.
        host.Tell(new VirtualTagActor.DependencyValueChanged("H.T", 99, DateTime.UtcNow));
        var evtTrue = alerts.FishForMessage<AlarmTransitionEvent>(
            e => e.AlarmId == "alm-hist-true" && e.TransitionKind == "Activated", Timeout);
        evtTrue.HistorizeToAveva.ShouldBe(true);

        // Activate the false-flag alarm.
        host.Tell(new VirtualTagActor.DependencyValueChanged("H.F", 99, DateTime.UtcNow));
        var evtFalse = alerts.FishForMessage<AlarmTransitionEvent>(
            e => e.AlarmId == "alm-hist-false" && e.TransitionKind == "Activated", Timeout);
        evtFalse.HistorizeToAveva.ShouldBe(false);
    }

    /// <summary>#478 — a Bad-quality dependency (no threshold crossing → no state transition) drives the
    /// scripted condition's Quality out of band: the host publishes an <see cref="OpcUaPublishActor.AlarmQualityUpdate"/>
    /// (Bad, Uns realm) and NO <c>/alerts</c> transition — quality is an annotation, not a state change.</summary>
    [Fact]
    public void Bad_quality_dependency_publishes_AlarmQualityUpdate_and_no_alerts()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T", threshold: 90) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Baseline is Good (unread inputs are not a quality signal — no load-time annotation). Drive the
        // input Bad with a value below threshold: the predicate freezes (no transition), but the worst-input
        // quality bucket moves Good→Bad → a QualityChanged annotation flows out of band.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 10, DateTime.UtcNow, OpcUaQuality.Bad));

        var q = publish.FishForMessage<OpcUaPublishActor.AlarmQualityUpdate>(m => m.Quality == OpcUaQuality.Bad, Timeout);
        q.AlarmNodeId.ShouldBe("alm-1");
        q.Realm.ShouldBe(AddressSpaceRealm.Uns);

        // A pure quality change is NOT a state transition: no /alerts row.
        alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
    }

    /// <summary>#478 — when a transition fires while an input is Uncertain, the projected full snapshot
    /// carries that worst-of-input quality (not a clobbered Good), so the condition node reflects that its
    /// state is derived from untrustworthy inputs.</summary>
    [Fact]
    public void Transition_snapshot_carries_worst_input_quality()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T", threshold: 90) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Above threshold (activates) but Uncertain quality — Uncertain is "ready", so the predicate runs.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow, OpcUaQuality.Uncertain));

        var state = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        state.AlarmNodeId.ShouldBe("alm-1");
        state.State.Quality.ShouldBe(OpcUaQuality.Uncertain);
    }

    /// <summary>OneShotShelve transition carries the operator's identity: an operator-driven OneShotShelve
    /// drives <c>OneShotShelveAsync</c> — the resulting <see cref="AlarmTransitionEvent"/>(<c>"Shelved"</c>)
    /// on the alerts topic must carry <c>User == cmd.User</c> (the acting operator), NOT the generic
    /// <c>"system"</c> sentinel. This verifies the <see cref="ScriptedAlarmHostActor"/>
    /// <c>TransitionUser</c> arm for <see cref="EmissionKind.Shelved"/>.</summary>
    [Fact]
    public void Shelved_transition_carries_operator_user()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(id: "alm-1", depRef: "M.T") }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Activate so there is something to shelve.
        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
        publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        alerts.FishForMessage<AlarmTransitionEvent>(e => e.TransitionKind == "Activated", Timeout);

        // Shelve via a one-shot shelve command — the host owns alm-1, so OneShotShelveAsync runs.
        host.Tell(new AlarmCommand(
            AlarmId: "alm-1", Operation: "OneShotShelve", User: "carol", Comment: null, UnshelveAtUtc: null));

        // The Shelved AlarmTransitionEvent must carry the operator's identity, not "system".
        var evt = alerts.FishForMessage<AlarmTransitionEvent>(e => e.TransitionKind == "Shelved", Timeout);
        evt.AlarmId.ShouldBe("alm-1");
        evt.User.ShouldBe("carol");   // operator — NOT "system"
    }

    /// <summary>Absent-node default-emit (A1): a <see cref="RedundancyStateChanged"/> snapshot that
    /// contains ONLY other nodes (the host's own <see cref="LocalNode"/> is absent) must leave the
    /// cached local role unchanged (null/unknown) — the host therefore defaults to emit, publishing
    /// the alerts transition AND writing the OPC UA node, exactly as in the boot-window case.</summary>
    [Fact]
    public void Redundancy_snapshot_without_local_node_leaves_role_unknown_and_emits()
    {
        var publish = CreateTestProbe();
        var mux = CreateTestProbe();
        var alerts = CreateTestProbe();
        SubscribeToAlerts(alerts);

        var (host, _) = Spawn(publish, mux, LocalNode);
        host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout); // load completed

        // Send a snapshot that mentions only a DIFFERENT node — LocalNode is absent.
        // The host cannot determine its own role from this snapshot, so the cached role
        // stays null (unknown) ⇒ treated as Primary ⇒ default-emit path.
        TellRedundancyRole(host, RedundancyRole.Primary, nodeId: new NodeId("some-other-node"));

        host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));

        // Both the OPC UA write AND the alerts publish must arrive (default-emit, role unknown).
        var state = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
        state.AlarmNodeId.ShouldBe("alm-1");

        var evt = alerts.ExpectMsg<AlarmTransitionEvent>(Timeout);
        evt.AlarmId.ShouldBe("alm-1");
        evt.TransitionKind.ShouldBe("Activated");
    }
}