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

using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.OtOpcUa.Cluster.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Health;

namespace ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;

/// <summary>
///     Single-threaded bridge between Akka messages and the OPC UA SDK address space. Hosted on
///     the pinned <c>opcua-synchronized-dispatcher</c> (HOCON) so the OPC UA SDK sees
///     only one thread per actor instance — its session/subscription locks expect strict
///     single-threaded access.
///
///     Address-space writes route through <see cref="IOpcUaAddressSpaceSink"/>; ServiceLevel
///     writes route through <see cref="IServiceLevelPublisher"/>. Production binds SDK-backed
///     implementations; dev/Mac/tests bind the Null* defaults so the actor stays decoupled from
///     <c>Opc.Ua.Server</c>. The remaining piece is wiring those bindings to a real
///     <c>StandardServer</c> address space — tracked as F10b.
/// </summary>
public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
{
    public const string DispatcherId = "opcua-synchronized-dispatcher";
    public const string RedundancyStateTopic = "redundancy-state";

    /// <summary>Gets or sets the timer scheduler for the periodic DB-health refresh tick.</summary>
    public ITimerScheduler Timers { get; set; } = null!;

    /// <summary>Self-tick that drives the periodic Ask of the local <see cref="DbHealthProbeActor"/>
    /// for its cached status. Private singleton — the actor pumps this for itself, no external sender.</summary>
    private sealed class HealthTick { public static readonly HealthTick Instance = new(); private HealthTick() { } }

    /// <summary>A driver/vtag value update for a single materialised Variable node. v3 Batch 4: carries the
    /// node's <see cref="AddressSpaceRealm"/> so the sink resolves the right namespace — the driver fan-out
    /// posts one of these per registered NodeId (the raw node in <see cref="AddressSpaceRealm.Raw"/> and each
    /// referencing UNS node in <see cref="AddressSpaceRealm.Uns"/>) with identical value/status/timestamp.
    /// <see cref="Realm"/> defaults to <see cref="AddressSpaceRealm.Uns"/> so pre-v3 callers (VirtualTag
    /// equipment nodes) keep compiling; the driver fan-out passes it EXPLICITLY per node.
    /// <para><see cref="StatusCode"/> is the full OPC UA status code, sub-code included; the sink
    /// hands it to the SDK verbatim.</para></summary>
    public sealed record AttributeValueUpdate(string NodeId, object? Value, uint StatusCode, DateTime TimestampUtc, AddressSpaceRealm Realm);

    /// <summary>Batched form of <see cref="AttributeValueUpdate"/>: every fan-out write of one driver
    /// production burst in one message — the sink takes its lock once for the whole batch. Items carry
    /// their realm each (<see cref="ValueWrite"/>); status codes ride verbatim.</summary>
    public sealed record AttributeValueBatchUpdate(IReadOnlyList<ValueWrite> Items);

    /// <summary>Status-only batch: every named node keeps its CURRENT value and only its status +
    /// timestamp change (archreview 03/S15). The comms-loss fan-out sends one of these covering every
    /// NodeId a disconnected driver owns — the raw node and each referencing UNS node — so a client sees
    /// <see cref="OpcUaStatus.BadNoCommunication"/> over the last known value instead of a stale Good.
    /// <para>Deliberately a separate message from <see cref="AttributeValueBatchUpdate"/> rather than a
    /// flag on it: the host holds no value cache (v3's single-source design), so it has nothing to put in
    /// a <see cref="ValueWrite"/>, and only the node knows what value to retain.</para></summary>
    public sealed record AttributeStatusBatchUpdate(IReadOnlyList<StatusWrite> Items);

    /// <summary>Carries the full Part 9 condition state for a scripted alarm to the sink. The
    /// <paramref name="State"/> snapshot is the Commons projection the Runtime host maps from the engine's
    /// Core <c>AlarmConditionState</c> + severity/message — the actor stays decoupled from
    /// <c>Core.ScriptedAlarms</c>.</summary>
    /// <param name="AlarmNodeId">The alarm node id (== ScriptedAlarmId for scripted conditions; == RawPath for
    /// v3 native raw conditions).</param>
    /// <param name="State">The full condition state to project onto the node.</param>
    /// <param name="TimestampUtc">The source timestamp of the transition in UTC.</param>
    /// <param name="Realm">The namespace realm the condition lives in — <see cref="AddressSpaceRealm.Uns"/> for
    /// scripted alarms (default), <see cref="AddressSpaceRealm.Raw"/> for v3 native raw conditions.</param>
    public sealed record AlarmStateUpdate(string AlarmNodeId, AlarmConditionSnapshot State, DateTime TimestampUtc, AddressSpaceRealm Realm);
    /// <summary>Annotate a materialised condition's source-data Quality out of band from any alarm
    /// transition (the driver-connectivity path: comms lost → <see cref="OpcUaStatus.BadNoCommunication"/>,
    /// restored → <see cref="OpcUaStatus.Good"/>). Routed to <see cref="IOpcUaAddressSpaceSink.WriteAlarmQuality"/>,
    /// which sets ONLY Quality and fires one Part 9 event on a status-code change.</summary>
    /// <param name="AlarmNodeId">The condition node id (RawPath for a native alarm).</param>
    /// <param name="StatusCode">The source-data status code to annotate, sub-code included.</param>
    /// <param name="TimestampUtc">The connectivity transition timestamp in UTC.</param>
    /// <param name="Realm">The namespace realm the condition lives in (<see cref="AddressSpaceRealm.Raw"/> for native).</param>
    public sealed record AlarmQualityUpdate(string AlarmNodeId, uint StatusCode, DateTime TimestampUtc, AddressSpaceRealm Realm);
    /// <summary>
    ///     Triggers an address-space rebuild of the artifact carried in <paramref name="Artifact"/>.
    ///     <paramref name="DeploymentId"/> is the deployment just applied by the host (echoed in the reply and
    ///     the logs). Review 02 P5: the ConfigDb fallback for a message carrying no bytes is gone — it had no
    ///     production producer and would have blocked the pinned OPC UA thread on central SQL — so a message
    ///     without bytes is "no answer" and holds the served address space.
    /// </summary>
    /// <param name="Correlation">Correlation id for tracing this rebuild.</param>
    /// <param name="DeploymentId">The applied deployment whose artifact this is, or null for a legacy caller.</param>
    /// <param name="Artifact">
    ///     The artifact bytes. Every producer (the deploy path, boot-from-cache, FetchAndCache) has them in
    ///     hand; null or empty is treated as no answer.
    /// </param>
    /// <param name="ReportOutcome">
    ///     when <see langword="true"/>, the actor replies to the sender with a
    ///     <see cref="RebuildCompleted"/> once the apply outcome is known (clean or degraded), on EVERY path
    ///     through the handler. <c>DriverHostActor</c> sets it on the deploy path and sends the deployment's
    ///     ack only after that reply, so a degraded apply can no longer seal a deployment green. Off for the
    ///     boot/restore paths, which re-ack nothing.
    /// </param>
    public sealed record RebuildAddressSpace(
        CorrelationId Correlation, DeploymentId? DeploymentId = null, byte[]? Artifact = null,
        bool ReportOutcome = false);

    /// <summary>
    ///     the reply to a <see cref="RebuildAddressSpace"/> carrying
    ///     <see cref="RebuildAddressSpace.ReportOutcome"/>: whether the address-space apply was clean.
    /// </summary>
    /// <param name="Correlation">The rebuild's correlation id, echoed.</param>
    /// <param name="DeploymentId">The rebuild's deployment id, echoed (null for a legacy caller).</param>
    /// <param name="Degraded">
    ///     <see langword="true"/> when the apply swallowed per-node sink faults, its rebuild failed, the
    ///     pipeline threw, or there was no usable artifact — i.e. the served address space may be partial or
    ///     stale. An empty plan (nothing changed) is clean.
    /// </param>
    /// <param name="DegradedReason">A short operator-readable reason; null when clean.</param>
    public sealed record RebuildCompleted(
        CorrelationId Correlation, DeploymentId? DeploymentId, bool Degraded, string? DegradedReason);

    public sealed record ServiceLevelChanged(byte ServiceLevel);

    /// <summary>
    ///     Review 02 P5 — a rebuild whose artifact parse, cluster-scoped composition and plan were computed
    ///     OFF the pinned OPC UA thread (a <c>Task.Run</c> piped back to <c>Self</c>), ready for the sink passes,
    ///     which are the only part that must run on this actor's thread. Carries the <see cref="Request"/> and
    ///     the original <see cref="Requester"/> so the <see cref="RebuildCompleted"/> reply reaches whoever asked.
    /// </summary>
    /// <para><see cref="Incarnation"/> stamps the actor instance that started the plan: a <c>PipeTo(Self)</c>
    ///     outlives a restart (the ActorRef survives it), and a plan computed against a previous instance's
    ///     <see cref="_lastApplied"/> must not be applied by the next one. <see cref="AuthorityError"/> carries a
    ///     data-plane-authority parse failure, which never fails the address-space apply (review fix I2).</para>
    /// </summary>
    private sealed record PlanReady(
        object Incarnation, RebuildAddressSpace Request, IActorRef Requester, byte[] Artifact,
        DataPlaneAuthorityProjection? Authority, Exception? AuthorityError,
        AddressSpaceComposition Composition, AddressSpacePlan Plan, IReadOnlyList<string> Inconsistencies);

    /// <summary>The off-thread parse / plan step threw (unreadable bytes, a composer fault).</summary>
    private sealed record PlanFailed(object Incarnation, RebuildAddressSpace Request, IActorRef Requester, Exception Error);

    /// <summary>Test seams for the off-thread plan step. Production passes none.</summary>
    internal sealed class TestHooks
    {
        /// <summary>Runs on the thread pool at the start of every plan step (a test blocks it to hold a plan in flight).</summary>
        internal Action? BeforePlan { get; init; }

        /// <summary>Replaces <see cref="ParsedDeploymentArtifact.ParseDataPlaneAuthority"/> (a test makes it throw).</summary>
        internal Func<ParsedDeploymentArtifact, string, DataPlaneAuthorityProjection>? ParseAuthority { get; init; }
    }

    /// <summary>The default bound on <see cref="_deferred"/>: distinct (realm, id) keys held while a plan is in flight.</summary>
    internal const int DefaultDeferredWriteCapacity = 250_000;

    /// <summary>Identity of THIS actor instance (see <see cref="PlanReady.Incarnation"/>).</summary>
    private readonly object _incarnation = new();
    private readonly TestHooks? _hooks;

    /// <summary>
    ///     Review fix C1 — the latest value / status / alarm write per (realm, id) received while a rebuild's plan is
    ///     in flight (or queued), replayed through the sink after each apply. Writes are non-creating (review 01 L1),
    ///     so a driver's first publish for a node the in-flight plan ADDS used to be skipped — and a change-driven
    ///     driver (or a poll group whose last-value cache already holds it) never sends it again, leaving the node
    ///     <c>BadWaitingForInitialData</c> indefinitely. On a full rebuild the same holds for every node: the
    ///     re-subscribe's first values land on old nodes the apply then discards.
    ///     <para>Writes are still applied IMMEDIATELY (existing nodes keep updating during the plan); the buffer
    ///     only remembers each key's latest so the replay can land it on a node the apply created. Replaying onto a
    ///     node that already holds it is inert: the same value/status/timestamp is filtered by every monitored
    ///     item's data-change filter, and an alarm write is delta-gated against the node's current state. That is
    ///     also why alarms coalesce to the latest state per id rather than replaying every transition: replaying an
    ///     older transition onto a live condition would move it backwards and fire spurious events.</para>
    /// </summary>
    private readonly DeferredWrites _deferred;

    /// <summary>
    ///     Review fix round 2 (N2 / M1) — WHICH deferred writes the next replay may land: only the ones whose live write
    ///     could not have, i.e. onto nodes the apply just CREATED. A node the apply kept took the live write already,
    ///     and replaying it is not inert: a client changed it in the window (an SDK-applied Acknowledge / Confirm /
    ///     Shelve on a condition, an optimistic Write on a variable) and the replay would roll that back and fire
    ///     spurious Part 9 events. Set by <see cref="ApplyPlan"/>, consumed by <see cref="FinishRebuild"/>.
    ///     <list type="bullet">
    ///         <item>Variables: after a FULL rebuild every one (each was recreated, its live write landed on the
    ///             discarded node); otherwise exactly those the apply asked the sink to ensure (the applier records them).</item>
    ///         <item>Variables an in-place attribute update RE-SHAPED (review N4): the node manager reset their value to
    ///             BadWaitingForInitialData, so the window's value is replayed — but only once fitted to the new shape
    ///             (<see cref="VariableShape.TryFit"/>); a value that does not fit is dropped and the node left waiting.
    ///             The raw node and its UNS mirror are re-shaped by the same edit, so both get the same treatment.</item>
    ///         <item>Conditions: those in the applied composition but not the previous one; after a full rebuild ALSO
    ///             every kept SCRIPTED condition (review N3) — the node manager carries only NATIVE conditions' state over
    ///             a rebuild, so a scripted one is recreated fresh and needs the window's state. A kept native condition
    ///             is restored from the carry-over (which already holds the live write and any SDK ack), so it is not
    ///             replayed. Accepted cost on the scripted path: an SDK ack on the discarded instance in the window is
    ///             lost with it; the engine's own ack emission (a scripted ack routes to the engine) restores it.</item>
    ///         <item>Known gaps (degraded path only, review m1 / m2): a degraded non-rebuild apply does not advance
    ///             <c>_lastApplied</c>, so the next apply counts its conditions as added again and may replay onto them;
    ///             and the empty-plan repair of a degraded composition scopes no condition (the composition is
    ///             unchanged), so a condition the repair newly places gets no window state — writes to it between the
    ///             two applies were already dropped before this replay existed.</item>
    ///     </list>
    ///     (Precise "did the live write land" would need the sink to report it; the production sink chain runs through
    ///     Commons' DeferredAddressSpaceSink, outside this lane — follow-up F7.)
    /// </summary>
    private ReplayScope _replayScope = ReplayScope.Nothing;

    private sealed record ReplayScope(
        bool AllVariables,
        IReadOnlySet<(AddressSpaceRealm, string)> Variables,
        IReadOnlyDictionary<(AddressSpaceRealm, string), VariableShape> Reshaped,
        IReadOnlySet<(AddressSpaceRealm, string)> Conditions)
    {
        private static readonly IReadOnlySet<(AddressSpaceRealm, string)> None = new HashSet<(AddressSpaceRealm, string)>();
        public static readonly ReplayScope Nothing =
            new(false, None, new Dictionary<(AddressSpaceRealm, string), VariableShape>(), None);
    }

    /// <summary>True while a rebuild's parse / plan step is running off-thread. Rebuilds are applied strictly in
    /// arrival order — the plan is a diff against <see cref="_lastApplied"/>, which only the apply advances — so a
    /// rebuild arriving meanwhile waits in <see cref="_pendingRebuilds"/>; value, status and alarm writes do not — they
    /// are applied at once AND remembered in <see cref="_deferred"/> for replay after the apply.</summary>
    private bool _planInFlight;
    private readonly Queue<(RebuildAddressSpace Request, IActorRef Requester)> _pendingRebuilds = new();

    private readonly IOpcUaAddressSpaceSink _sink;
    private readonly IServiceLevelPublisher _serviceLevel;
    private readonly bool _subscribeRedundancyTopic;
    private readonly NodeId? _localNode;
    private readonly AddressSpaceApplier? _applier;
    /// <summary>Receives the API keys + ACL grants carried by each applied artifact, so
    /// OPC UA session activation can authenticate. Null when nothing consumes it (tests, and any node
    /// that is not serving an OPC UA data plane).</summary>
    private readonly IDataPlaneAuthoritySink? _dataPlaneAuthority;
    private readonly IActorRef? _dbHealthProbe;
    /// <summary>Per-cluster mesh Phase 4: this is a driver-only node with NO ConfigDb (LocalDb is its
    /// config store). There is no DB to probe, so the ServiceLevel DB-reachability input is retired —
    /// <c>DbReachable</c> is fed constant <c>true</c> and only snapshot staleness can demote. See
    /// <see cref="RecomputeServiceLevel"/>.</summary>
    private readonly bool _dbLess;
    /// <summary>This node's OPC UA endpoint lifecycle fact (lmxopcua#762). Null = no fact (harness) = listening.</summary>
    private readonly IOpcUaEndpointState? _endpointState;
    private readonly TimeSpan _staleWindow;
    private readonly TimeSpan _probeFreshnessWindow;
    private readonly TimeSpan _healthTickInterval;
    private readonly Func<DateTime> _utcNow;
    private readonly Akka.Cluster.Cluster _cluster = Akka.Cluster.Cluster.Get(Context.System);
    private readonly ILoggingAdapter _log = Context.GetLogger();

    private int _writes;
    private byte _lastServiceLevel;
    private bool _publishedAtLeastOnce;
    /// <summary>True once a real composition has been applied, i.e. there is a served address space worth
    /// protecting. Distinguishes "the artifact went missing under a running server" (warn, and
    /// hold what is materialised) from "nothing has been deployed here yet" (a quiet no-op at boot).</summary>
    private bool _hasAppliedComposition;
    /// <summary>
    ///     <see cref="_lastApplied"/> was advanced by a DEGRADED rebuild (a rebuild advances
    ///     even with swallowed failures, see <see cref="HandleRebuildCore"/>). A re-deploy of the same revision
    ///     then diffs to an EMPTY plan; without this flag that empty plan reported clean, so the "re-deploy
    ///     re-applies" repair path sealed green over an address space still known to be partial. While set, an
    ///     empty plan re-materialises the whole composition (idempotent ensures) and reports clean only when
    ///     that pass is itself clean.
    /// </summary>
    private bool _lastAppliedDegraded;
    private DbHealthProbeActor.DbHealthStatus? _lastDbHealth;
    private RedundancyStateChanged? _lastSnapshot;
    private (bool Ok, DateTime At)? _probeAboutMe;
    private AddressSpaceComposition _lastApplied = new(
        Array.Empty<UnsAreaProjection>(),
        Array.Empty<UnsLineProjection>(),
        Array.Empty<EquipmentNode>(),
        Array.Empty<DriverInstancePlan>(),
        Array.Empty<ScriptedAlarmPlan>());

    /// <summary>Gets the number of writes performed.</summary>
    public int WriteCount => _writes;
    /// <summary>Gets the last published service level.</summary>
    public byte LastServiceLevel => _lastServiceLevel;

    /// <summary>Production Props — pins the OPC UA dispatcher + subscribes to the
    /// <c>redundancy-state</c> DPS topic so cluster transitions drive the local ServiceLevel
    /// publish path. When <paramref name="applier"/> is supplied, <see cref="RebuildAddressSpace"/>
    /// diffs + applies the artifact bytes it carries through the sink (<paramref name="dbFactory"/> is
    /// accepted for wiring compatibility only — review 02 P5 removed the ConfigDb fallback). When <paramref name="dbHealthProbe"/> is supplied the local
    /// ServiceLevel is computed via <see cref="ServiceLevelCalculator"/> from real DB-health +
    /// staleness + role-leader inputs; otherwise the legacy role-only switch is used.</summary>
    /// <param name="sink">The OPC UA address space sink.</param>
    /// <param name="serviceLevel">The service level publisher.</param>
    /// <param name="localNode">The local cluster node ID.</param>
    /// <param name="dbFactory">The optional database context factory.</param>
    /// <param name="applier">The optional Phase 7 applier.</param>
    /// <param name="dbHealthProbe">The optional <see cref="DbHealthProbeActor"/> ref; when null the
    /// legacy role-only ServiceLevel seam is used until a <see cref="DbHealthProbeActor.DbHealthStatus"/> arrives.</param>
    /// <param name="staleWindow">The window beyond which a DB-health sample or redundancy snapshot is
    /// considered stale; defaults to 30 seconds.</param>
    /// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
    /// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
    /// defaults to 30 seconds.</param>
    /// <param name="healthTickInterval">The period between freshness ticks; defaults to 5 seconds.
    /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness.
    /// Legacy nodes without a probe do not start a timer.</param>
    /// <param name="utcNow">The UTC clock used for freshness; defaults to <see cref="DateTime.UtcNow"/>.</param>
    /// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
    /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
    /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
    /// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param>
    /// <param name="endpointState">This node's OPC UA endpoint lifecycle fact (lmxopcua#762). A
    /// <see cref="OpcUaEndpointStatus.Failed"/> endpoint computes ServiceLevel 0. Null (harnesses, a node with no
    /// hosted SDK) is treated as listening.</param>
    /// <returns>The configured <see cref="Props"/> for creating this actor, pinned to the
    /// <see cref="DispatcherId"/> dispatcher.</returns>
    public static Props Props(
        IOpcUaAddressSpaceSink? sink = null,
        IServiceLevelPublisher? serviceLevel = null,
        NodeId? localNode = null,
        IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory = null,
        AddressSpaceApplier? applier = null,
        IActorRef? dbHealthProbe = null,
        TimeSpan? staleWindow = null,
        TimeSpan? probeFreshnessWindow = null,
        TimeSpan? healthTickInterval = null,
        bool dbLess = false,
        IDataPlaneAuthoritySink? dataPlaneAuthority = null,
        Func<DateTime>? utcNow = null,
        IOpcUaEndpointState? endpointState = null) =>
        Akka.Actor.Props.Create(() => new OpcUaPublishActor(
            sink ?? NullOpcUaAddressSpaceSink.Instance,
            serviceLevel ?? NullServiceLevelPublisher.Instance,
            subscribeRedundancyTopic: true,
            localNode,
            dbFactory,
            applier,
            dbHealthProbe,
            staleWindow,
            probeFreshnessWindow,
            healthTickInterval,
            dbLess,
            dataPlaneAuthority,
            utcNow,
            endpointState)).WithDispatcher(DispatcherId);

    /// <summary>Test-only Props that omits the pinned-dispatcher requirement and skips the
    /// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster.</summary>
    /// <param name="sink">The OPC UA address space sink.</param>
    /// <param name="serviceLevel">The service level publisher.</param>
    /// <param name="subscribeRedundancyTopic">Whether to subscribe to the redundancy topic.</param>
    /// <param name="localNode">The local cluster node ID.</param>
    /// <param name="dbFactory">The optional database context factory.</param>
    /// <param name="applier">The optional Phase 7 applier.</param>
    /// <param name="dbHealthProbe">The optional <see cref="DbHealthProbeActor"/> ref; when null the
    /// legacy role-only ServiceLevel seam is used until a <see cref="DbHealthProbeActor.DbHealthStatus"/> arrives.</param>
    /// <param name="staleWindow">The window beyond which a DB-health sample or redundancy snapshot is
    /// considered stale; defaults to 30 seconds.</param>
    /// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
    /// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
    /// defaults to 30 seconds.</param>
    /// <param name="healthTickInterval">The period between freshness ticks; defaults to 5 seconds.
    /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness.
    /// Legacy nodes without a probe do not start a timer.</param>
    /// <param name="utcNow">The UTC clock used for freshness; defaults to <see cref="DateTime.UtcNow"/>.</param>
    /// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
    /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
    /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
    /// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param>
    /// <param name="endpointState">This node's OPC UA endpoint lifecycle fact (lmxopcua#762). A
    /// <see cref="OpcUaEndpointStatus.Failed"/> endpoint computes ServiceLevel 0. Null (harnesses, a node with no
    /// hosted SDK) is treated as listening.</param>
    /// <returns>The configured <see cref="Props"/> for creating this actor in tests, without dispatcher
    /// pinning or the DPS subscribe.</returns>
    public static Props PropsForTests(
        IOpcUaAddressSpaceSink? sink = null,
        IServiceLevelPublisher? serviceLevel = null,
        bool subscribeRedundancyTopic = false,
        NodeId? localNode = null,
        IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory = null,
        AddressSpaceApplier? applier = null,
        IActorRef? dbHealthProbe = null,
        TimeSpan? staleWindow = null,
        TimeSpan? probeFreshnessWindow = null,
        TimeSpan? healthTickInterval = null,
        bool dbLess = false,
        IDataPlaneAuthoritySink? dataPlaneAuthority = null,
        Func<DateTime>? utcNow = null,
        IOpcUaEndpointState? endpointState = null) =>
        Akka.Actor.Props.Create(() => new OpcUaPublishActor(
            sink ?? NullOpcUaAddressSpaceSink.Instance,
            serviceLevel ?? NullServiceLevelPublisher.Instance,
            subscribeRedundancyTopic,
            localNode,
            dbFactory,
            applier,
            dbHealthProbe,
            staleWindow,
            probeFreshnessWindow,
            healthTickInterval,
            dbLess,
            dataPlaneAuthority,
            utcNow,
            endpointState));

    /// <summary>Test-only Props exposing the deferred-write bound and the plan-step seams.</summary>
    internal static Props PropsForTests(
        IOpcUaAddressSpaceSink sink,
        AddressSpaceApplier? applier,
        TestHooks? hooks,
        int deferredWriteCapacity = DefaultDeferredWriteCapacity,
        NodeId? localNode = null,
        IDataPlaneAuthoritySink? dataPlaneAuthority = null) =>
        // CreateBy, not Create(expression): Akka's expression path activates through a PUBLIC constructor, and the
        // seam-carrying constructor is internal so the test seams stay off the public API (review M3).
        Akka.Actor.Props.CreateBy(new TestProducer(() => new OpcUaPublishActor(
            sink, NullServiceLevelPublisher.Instance, false, localNode, null, applier, null, null, null, null,
            false, dataPlaneAuthority, null, deferredWriteCapacity, hooks, endpointState: null)));

    /// <summary>Produces the actor through its internal (seam-carrying) constructor, for <see cref="PropsForTests(IOpcUaAddressSpaceSink, AddressSpaceApplier?, TestHooks?, int, NodeId?, IDataPlaneAuthoritySink?)"/>.</summary>
    private sealed class TestProducer(Func<OpcUaPublishActor> create) : IIndirectActorProducer
    {
        public Type ActorType => typeof(OpcUaPublishActor);
        public ActorBase Produce() => create();
        public void Release(ActorBase actor) { }
    }

    /// <summary>Initializes a new instance of the <see cref="OpcUaPublishActor"/> class.</summary>
    /// <param name="sink">The OPC UA address space sink.</param>
    /// <param name="serviceLevel">The service level publisher.</param>
    /// <param name="subscribeRedundancyTopic">Whether to subscribe to the redundancy topic.</param>
    /// <param name="localNode">The local cluster node ID.</param>
    /// <param name="dbFactory">The optional database context factory.</param>
    /// <param name="applier">The optional Phase 7 applier.</param>
    /// <param name="dbHealthProbe">The optional <see cref="DbHealthProbeActor"/> ref; when null the
    /// legacy role-only ServiceLevel seam is used until a <see cref="DbHealthProbeActor.DbHealthStatus"/> arrives.</param>
    /// <param name="staleWindow">The window beyond which a DB-health sample or redundancy snapshot is
    /// considered stale; defaults to 30 seconds.</param>
    /// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
    /// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
    /// defaults to 30 seconds.</param>
    /// <param name="healthTickInterval">The period between freshness ticks; defaults to 5 seconds.
    /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness.
    /// Legacy nodes without a probe do not start a timer.</param>
    /// <param name="utcNow">The UTC clock used for freshness; defaults to <see cref="DateTime.UtcNow"/>.</param>
    /// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
    /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
    /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
    /// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param>
    /// <param name="endpointState">This node's OPC UA endpoint lifecycle fact (lmxopcua#762); null is treated as listening.</param>
    public OpcUaPublishActor(
        IOpcUaAddressSpaceSink sink,
        IServiceLevelPublisher serviceLevel,
        bool subscribeRedundancyTopic,
        NodeId? localNode,
        IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory = null,
        AddressSpaceApplier? applier = null,
        IActorRef? dbHealthProbe = null,
        TimeSpan? staleWindow = null,
        TimeSpan? probeFreshnessWindow = null,
        TimeSpan? healthTickInterval = null,
        bool dbLess = false,
        IDataPlaneAuthoritySink? dataPlaneAuthority = null,
        Func<DateTime>? utcNow = null,
        IOpcUaEndpointState? endpointState = null)
        : this(sink, serviceLevel, subscribeRedundancyTopic, localNode, dbFactory, applier, dbHealthProbe, staleWindow,
            probeFreshnessWindow, healthTickInterval, dbLess, dataPlaneAuthority, utcNow, DefaultDeferredWriteCapacity, hooks: null,
            endpointState)
    {
    }

    /// <summary>The full constructor: also the deferred-write bound and the test seams (internal — tests only).</summary>
    /// <param name="deferredWriteCapacity">The bound on distinct nodes whose writes are held for replay while a rebuild plan is in flight.</param>
    /// <param name="hooks">Test seams for the off-thread plan step; null in production.</param>
    /// <param name="endpointState">This node's OPC UA endpoint lifecycle fact (lmxopcua#762); null is treated as listening.</param>
    internal OpcUaPublishActor(
        IOpcUaAddressSpaceSink sink,
        IServiceLevelPublisher serviceLevel,
        bool subscribeRedundancyTopic,
        NodeId? localNode,
        IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
        AddressSpaceApplier? applier,
        IActorRef? dbHealthProbe,
        TimeSpan? staleWindow,
        TimeSpan? probeFreshnessWindow,
        TimeSpan? healthTickInterval,
        bool dbLess,
        IDataPlaneAuthoritySink? dataPlaneAuthority,
        Func<DateTime>? utcNow,
        int deferredWriteCapacity,
        TestHooks? hooks,
        IOpcUaEndpointState? endpointState)
    {
        _deferred = new DeferredWrites(Math.Max(1, deferredWriteCapacity));
        _hooks = hooks;
        _sink = sink;
        _dataPlaneAuthority = dataPlaneAuthority;
        _serviceLevel = serviceLevel;
        _subscribeRedundancyTopic = subscribeRedundancyTopic;
        _localNode = localNode;
        // dbFactory is accepted for wiring compatibility only (review 02 P5): the bytes-less ConfigDb read the
        // rebuild used to fall back to is gone — every producer carries the artifact bytes in the message.
        _ = dbFactory;
        _applier = applier;
        _dbHealthProbe = dbHealthProbe;
        _dbLess = dbLess;
        _endpointState = endpointState;
        _staleWindow = staleWindow ?? TimeSpan.FromSeconds(30);
        _probeFreshnessWindow = probeFreshnessWindow ?? TimeSpan.FromSeconds(30);
        _healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5);
        _utcNow = utcNow ?? (() => DateTime.UtcNow);

        Receive<AttributeValueUpdate>(HandleAttributeUpdate);
        Receive<AttributeValueBatchUpdate>(HandleAttributeBatchUpdate);
        Receive<AttributeStatusBatchUpdate>(HandleAttributeStatusBatchUpdate);
        Receive<AlarmStateUpdate>(HandleAlarmUpdate);
        Receive<AlarmQualityUpdate>(HandleAlarmQualityUpdate);
        Receive<RebuildAddressSpace>(HandleRebuild);
        Receive<PlanReady>(HandlePlanReady);
        Receive<PlanFailed>(HandlePlanFailed);
        Receive<ServiceLevelChanged>(HandleServiceLevelChanged);
        Receive<RedundancyStateChanged>(HandleRedundancyStateChanged);
        Receive<DbHealthProbeActor.DbHealthStatus>(HandleDbHealthStatus);
        Receive<HealthTick>(_ => OnHealthTick());
        Receive<PeerOpcUaProbeActor.OpcUaProbeResult>(HandlePeerProbe);
        Receive<SubscribeAck>(_ => { /* PubSub ack */ });
    }

    /// <summary>
    ///     Review M5 — a restart discards <see cref="_pendingRebuilds"/>; answer each queued requester (degraded) rather
    ///     than leave it to the deploy timeout. The in-flight plan's requester is answered when its plan pipes back to
    ///     the new instance (<see cref="IsFromAnotherIncarnation"/>).
    /// </summary>
    /// <inheritdoc />
    protected override void PreRestart(Exception reason, object message)
    {
        while (_pendingRebuilds.TryDequeue(out var queued))
        {
            if (!queued.Request.ReportOutcome) continue;
            queued.Requester.Tell(new RebuildCompleted(queued.Request.Correlation, queued.Request.DeploymentId, true,
                "the address-space publisher restarted before this rebuild was applied; it was not applied"));
        }

        base.PreRestart(reason, message);
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        if (_subscribeRedundancyTopic)
        {
            DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(RedundancyStateTopic, Self));
        }

        // DB-backed nodes refresh their cached probe status; DB-less nodes still need periodic
        // snapshot/probe-verdict age checks even when no new messages arrive. Legacy nodes keep
        // their role-only behavior without a timer.
        if (_dbLess || _dbHealthProbe is not null)
        {
            Self.Tell(HealthTick.Instance);                                        // immediate first refresh
            Timers.StartPeriodicTimer("db-health", HealthTick.Instance, _healthTickInterval);
        }
    }

    private void HandleAttributeUpdate(AttributeValueUpdate msg)
    {
        try
        {
            if (PlanPipelineBusy) DeferValue(new ValueWrite(msg.NodeId, msg.Value, msg.StatusCode, msg.TimestampUtc, msg.Realm));
            _sink.WriteValue(msg.NodeId, msg.Value, msg.StatusCode, msg.TimestampUtc, msg.Realm);
            Interlocked.Increment(ref _writes);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "value"));
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: sink.WriteValue threw for {Node}", msg.NodeId);
        }
    }

    /// <summary>Mirrors <see cref="HandleAttributeUpdate"/> for a whole burst: ONE
    /// <see cref="IOpcUaAddressSpaceSink.WriteValues"/> call, but the write tally and the
    /// <c>otopcua.opcua.sink.write</c> meter still advance PER ITEM — batching is a transport change, so
    /// dashboards must not drop by the batch factor.
    /// <para>A <see cref="ValueWriteBatchException"/> is a PARTIAL failure: the sink applied every item it
    /// could and is telling us which one it couldn't. So it is caught ahead of the general catch to (a) name
    /// the offending node in the log — a bare "the batch threw" cannot be acted on — and (b) still count the
    /// items that DID land, keeping the per-item counting invariant honest instead of silently under-reporting
    /// a whole batch because one item was bad.</para></summary>
    private void HandleAttributeBatchUpdate(AttributeValueBatchUpdate msg)
    {
        if (PlanPipelineBusy)
            foreach (var item in msg.Items) DeferValue(item);
        try
        {
            _sink.WriteValues(msg.Items);
            Interlocked.Add(ref _writes, msg.Items.Count);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair<string, object?>("kind", "value"));
        }
        catch (ValueWriteBatchException ex)
        {
            _log.Warning(ex,
                "OpcUaPublish: sink.WriteValues partially failed for a {Count}-item batch — {Failed} failed " +
                "(first: {Node}), {Succeeded} applied",
                msg.Items.Count, ex.FailedCount, ex.FirstFailedNodeId, ex.SucceededCount);
            Interlocked.Add(ref _writes, ex.SucceededCount);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(ex.SucceededCount, new KeyValuePair<string, object?>("kind", "value"));
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: sink.WriteValues threw for a {Count}-item batch", msg.Items.Count);
        }
    }

    /// <summary>Routes a status-only batch to <see cref="IOpcUaAddressSpaceSink.WriteStatuses"/> — the
    /// comms-loss annotation (03/S15). Same shape as <see cref="HandleAttributeBatchUpdate"/>: metered per
    /// ITEM (batching is a transport change, so dashboards must not drop by the batch factor), a
    /// <see cref="ValueWriteBatchException"/> is a partial failure that still counts what landed and names
    /// the offender, and any other throw is swallowed so a sink fault cannot restart the publish actor and
    /// drop live traffic. Metered under <c>kind:"status"</c> so a status annotation is separable from a
    /// value write on the same counter.</summary>
    private void HandleAttributeStatusBatchUpdate(AttributeStatusBatchUpdate msg)
    {
        if (PlanPipelineBusy)
            foreach (var item in msg.Items) DeferStatus(item);
        try
        {
            _sink.WriteStatuses(msg.Items);
            Interlocked.Add(ref _writes, msg.Items.Count);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair<string, object?>("kind", "status"));
        }
        catch (ValueWriteBatchException ex)
        {
            _log.Warning(ex,
                "OpcUaPublish: sink.WriteStatuses partially failed for a {Count}-item batch — {Failed} failed " +
                "(first: {Node}), {Succeeded} applied",
                msg.Items.Count, ex.FailedCount, ex.FirstFailedNodeId, ex.SucceededCount);
            Interlocked.Add(ref _writes, ex.SucceededCount);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(ex.SucceededCount, new KeyValuePair<string, object?>("kind", "status"));
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: sink.WriteStatuses threw for a {Count}-item batch", msg.Items.Count);
        }
    }

    private void HandleAlarmUpdate(AlarmStateUpdate msg)
    {
        if (PlanPipelineBusy) DeferAlarm(msg);
        try
        {
            _sink.WriteAlarmCondition(msg.AlarmNodeId, msg.State, msg.TimestampUtc, msg.Realm);
            Interlocked.Increment(ref _writes);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "alarm"));
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: sink.WriteAlarmCondition threw for {Node}", msg.AlarmNodeId);
        }
    }

    private void HandleAlarmQualityUpdate(AlarmQualityUpdate msg)
    {
        if (PlanPipelineBusy) DeferAlarmQuality(msg);
        try
        {
            _sink.WriteAlarmQuality(msg.AlarmNodeId, msg.StatusCode, msg.TimestampUtc, msg.Realm);
            Interlocked.Increment(ref _writes);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "alarm-quality"));
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: sink.WriteAlarmQuality threw for {Node}", msg.AlarmNodeId);
        }
    }

    private void HandleRebuild(RebuildAddressSpace msg) => StartRebuild(msg, Sender);

    /// <summary>
    ///     The entry half of a rebuild, on the actor thread. The raw-sink fallback (no applier) and the
    ///     no-artifact case complete synchronously; otherwise the artifact parse, the cluster-scoped composition
    ///     and the plan — the O(artifact) work that used to serialise every live value batch and (because the
    ///     node manager's lock is the SDK's) every client Publish behind it — run on the thread pool and come
    ///     back as a <see cref="PlanReady"/>. Only the sink materialisation (<see cref="ApplyPlan"/>) runs here.
    /// </summary>
    private void StartRebuild(RebuildAddressSpace msg, IActorRef requester)
    {
        // Strict arrival order for EVERY rebuild, including the ones that complete synchronously (no applier, no
        // bytes): a bytes-less request arriving mid-plan used to answer before the plan ahead of it, so replies went
        // out of order. HandleRebuild is the only entry that can find the pipeline busy; the drain calls this only
        // when it is idle.
        if (_planInFlight)
        {
            _pendingRebuilds.Enqueue((msg, requester));
            return;
        }

        // Only a node with NO applier at all (the F10b/dev seam) falls back to the raw-sink rebuild — and that
        // fallback WIPES the served address space without re-materialising, so it must NEVER be reached by a
        // real driver node that has an applier (that node diff-and-applies from its in-hand bytes below).
        if (_applier is null)
        {
            // Every rebuild is traced, whichever path completes it: the applier path opens its span in ApplyPlan
            // (the actor-thread half), the two synchronous completions below open their own.
            using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan();
            span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString());
            (bool Degraded, string? Reason) outcome;
            try
            {
                _sink.RebuildAddressSpace();
                OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "rebuild"));
                outcome = (false, null);
            }
            catch (Exception ex)
            {
                _log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})", msg.Correlation);
                outcome = (true, $"address-space rebuild threw: {ex.Message}");
            }

            FinishRebuild(msg, requester, outcome);
            return;
        }

        // An artifact we could not obtain is NOT an empty configuration. Parsing zero bytes yields an empty
        // composition, which the planner diffs against the live one as a PureRemove and the applier then
        // faithfully tears down. Nothing legitimate produces a zero-length blob (an operator who really deletes
        // everything still deploys a JSON document with empty arrays), so the only honest reading of "no bytes"
        // is "no answer" — hold the last-known-good address space and let the next deploy (or the next boot)
        // supply a real one. Leaving _lastApplied intact means the retry diffs against what is materialised.
        if (msg.Artifact is null or { Length: 0 })
        {
            using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan();
            span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString());
            if (_hasAppliedComposition)
                _log.Warning(
                    "OpcUaPublish: no usable artifact for deployment {Id} (correlation={Correlation}) — KEEPING the currently served address space rather than tearing it down",
                    msg.DeploymentId, msg.Correlation);
            else
                _log.Debug(
                    "OpcUaPublish: no artifact to materialise yet (deployment={Id}, correlation={Correlation})",
                    msg.DeploymentId, msg.Correlation);
            FinishRebuild(msg, requester, (true, "no usable artifact reached the address-space apply; the previously served address space was kept"));
            return;
        }

        _planInFlight = true;
        var artifact = msg.Artifact;
        var baseComposition = _lastApplied;
        var nodeId = _localNode?.Value;
        var wantAuthority = _dataPlaneAuthority is not null;
        var incarnation = _incarnation;
        var hooks = _hooks;
        Task.Run(() =>
            {
                hooks?.BeforePlan?.Invoke();
                // Open() throws on bytes that are not valid JSON, where the static wrappers swallowed to an EMPTY
                // result — and an empty composition is not a benign no-op here (see the no-bytes guard above). The
                // throw lands in PlanFailed: logged at Error, _lastApplied untouched, last-known-good still served.
                using var parsed = DeploymentArtifact.Open(artifact);
                // The authority parse is isolated from the address-space plan (review fix I2): before the plan moved
                // off-thread a throw here was contained by FeedDataPlaneAuthority's own catch; it must not now
                // become PlanFailed and degrade an apply whose topology is perfectly readable.
                DataPlaneAuthorityProjection? authority = null;
                Exception? authorityError = null;
                if (wantAuthority)
                {
                    try
                    {
                        authority = hooks?.ParseAuthority is { } parse
                            ? parse(parsed, nodeId ?? string.Empty)
                            : parsed.ParseDataPlaneAuthority(nodeId ?? string.Empty);
                    }
                    catch (Exception ex)
                    {
                        authorityError = ex;
                    }
                }

                var inconsistencies = new List<string>();
                var composition = nodeId is not null
                    ? parsed.ParseComposition(nodeId, inconsistencies.Add)
                    : parsed.ParseComposition();
                var plan = AddressSpacePlanner.Compute(baseComposition, composition);
                return new PlanReady(incarnation, msg, requester, artifact, authority, authorityError, composition, plan, inconsistencies);
            })
            .PipeTo(Self, failure: ex => new PlanFailed(incarnation, msg, requester, ex));
    }

    private void HandlePlanReady(PlanReady ready)
    {
        if (IsFromAnotherIncarnation(ready.Incarnation, ready.Request, ready.Requester)) return;

        if (ready.AuthorityError is { } authorityError)
            _log.Error(authorityError,
                "OpcUaPublish: data-plane authority parse threw (correlation={Correlation}) — the address space is applied; the previously installed keys and grants stay in force",
                ready.Request.Correlation);

        (bool Degraded, string? Reason) outcome;
        try
        {
            outcome = ApplyPlan(ready);
        }
        finally
        {
            _planInFlight = false;
        }

        FinishRebuild(ready.Request, ready.Requester, outcome);
        AfterPlanSettled();
    }

    private void HandlePlanFailed(PlanFailed failed)
    {
        if (IsFromAnotherIncarnation(failed.Incarnation, failed.Request, failed.Requester)) return;

        _planInFlight = false;
        _log.Error(failed.Error, "OpcUaPublish: rebuild pipeline threw (correlation={Correlation})", failed.Request.Correlation);
        FinishRebuild(failed.Request, failed.Requester, (true, $"address-space apply threw: {failed.Error.Message}"));
        AfterPlanSettled();
    }

    /// <summary>A plan piped back to a RESTARTED instance (the ActorRef survives a restart) was computed against the
    /// previous instance's <see cref="_lastApplied"/>: never applied. Its requester is still owed an answer.</summary>
    private bool IsFromAnotherIncarnation(object incarnation, RebuildAddressSpace request, IActorRef requester)
    {
        if (ReferenceEquals(incarnation, _incarnation)) return false;
        _log.Warning(
            "OpcUaPublish: ignoring a rebuild plan started by a previous incarnation of this actor (correlation={Correlation})",
            request.Correlation);
        if (request.ReportOutcome)
            requester.Tell(new RebuildCompleted(request.Correlation, request.DeploymentId, true,
                "the address-space publisher restarted while this rebuild was being planned; it was not applied"));
        return true;
    }

    /// <summary>After an apply (or a failed plan) has been answered — its deferred writes already replayed by
    /// <see cref="FinishRebuild"/> — start what queued behind it, and drop the buffer once nothing is in flight or
    /// queued (a queued plan may add nodes the buffer names, so it is replayed again after that apply).</summary>
    private void AfterPlanSettled()
    {
        StartNextPendingRebuild();
        if (!PlanPipelineBusy) _deferred.Clear();
    }

    /// <summary>Starts queued rebuilds in arrival order until one goes off-thread (or the queue empties) — the
    /// synchronous completions (no bytes) answer in their turn.</summary>
    private void StartNextPendingRebuild()
    {
        while (!_planInFlight && _pendingRebuilds.TryDequeue(out var next))
            StartRebuild(next.Request, next.Requester);
    }

    /// <summary>True while a plan is computing or a rebuild is queued behind one.</summary>
    private bool PlanPipelineBusy => _planInFlight || _pendingRebuilds.Count > 0;

    private void DeferValue(in ValueWrite write)
    {
        if (!_deferred.Value(write)) WarnDeferredOverflow();
    }

    private void DeferStatus(in StatusWrite write)
    {
        if (!_deferred.Status(write)) WarnDeferredOverflow();
    }

    private void DeferAlarm(AlarmStateUpdate msg)
    {
        if (!_deferred.Alarm(msg)) WarnDeferredOverflow();
    }

    private void DeferAlarmQuality(AlarmQualityUpdate msg)
    {
        if (!_deferred.AlarmQuality(msg)) WarnDeferredOverflow();
    }

    private void WarnDeferredOverflow()
    {
        if (_deferred.Dropped == 1)
            _log.Warning(
                "OpcUaPublish: more than {Capacity} distinct nodes written while a rebuild plan is in flight — later new nodes are not held for replay (their first value lands only if the driver republishes it)",
                _deferred.Capacity);
    }

    /// <summary>Re-applies the deferred writes through the sink: values, then status-only annotations, then alarm
    /// states and alarm qualities. Each pass isolated like the live handlers; not re-metered (counted on arrival).</summary>
    private void ReplayDeferredWrites(ReplayScope scope)
    {
        if (_deferred.IsEmpty) return;
        var (allValues, allStatuses) = _deferred.VariableWrites(
            key => scope.AllVariables || scope.Variables.Contains(key) || scope.Reshaped.ContainsKey(key));
        var values = new List<ValueWrite>(allValues.Count);
        var statuses = new List<StatusWrite>(allStatuses.Count);
        var unfit = 0;
        foreach (var v in allValues)
        {
            if (!scope.Reshaped.TryGetValue((v.Realm, v.NodeId), out var shape)) { values.Add(v); continue; }
            if (shape.TryFit(v.Value, out var fitted)) { values.Add(v with { Value = fitted }); continue; }
            unfit++;
            _log.Debug(
                "OpcUaPublish: a value deferred for re-shaped {Realm} node {Node} does not fit its new shape ({DataType}, array={IsArray}) — not replayed; the node waits for the driver's next value",
                v.Realm, v.NodeId, shape.DataType, shape.IsArray);
        }

        foreach (var st in allStatuses)
        {
            // A status-only annotation onto a re-shaped (value-less) node would serve a null value under it; the
            // node's honest state is BadWaitingForInitialData until a value of the new shape arrives.
            if (scope.Reshaped.ContainsKey((st.Realm, st.NodeId))) { unfit++; continue; }
            statuses.Add(st);
        }
        if (values.Count > 0)
        {
            try { _sink.WriteValues(values); }
            catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying {Count} deferred value writes threw", values.Count); }
        }

        if (statuses.Count > 0)
        {
            try { _sink.WriteStatuses(statuses); }
            catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying {Count} deferred status writes threw", statuses.Count); }
        }

        var alarms = 0;
        foreach (var alarm in _deferred.Alarms)
        {
            if (!scope.Conditions.Contains((alarm.Realm, alarm.AlarmNodeId))) continue;
            alarms++;
            try { _sink.WriteAlarmCondition(alarm.AlarmNodeId, alarm.State, alarm.TimestampUtc, alarm.Realm); }
            catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying a deferred alarm write threw for {Node}", alarm.AlarmNodeId); }
        }

        var qualities = 0;
        foreach (var quality in _deferred.AlarmQualities)
        {
            if (!scope.Conditions.Contains((quality.Realm, quality.AlarmNodeId))) continue;
            qualities++;
            try { _sink.WriteAlarmQuality(quality.AlarmNodeId, quality.StatusCode, quality.TimestampUtc, quality.Realm); }
            catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying a deferred alarm-quality write threw for {Node}", quality.AlarmNodeId); }
        }

        _log.Debug(
            "OpcUaPublish: replayed {Values} value, {Statuses} status, {Alarms} alarm-state and {Qualities} alarm-quality writes (of {Held} held; {Unfit} not fitting a re-shaped node) deferred during a rebuild plan",
            values.Count, statuses.Count, alarms, qualities, _deferred.Count, unfit);
    }

    /// <summary>
    ///     The latest write per (realm, id) received while a plan is in flight — see <see cref="_deferred"/>. A value
    ///     followed by a status-only write for the same node composes to the value under the new status (exactly what
    ///     the two live writes leave on the node). Bounded by <see cref="Capacity"/> distinct keys across all kinds;
    ///     beyond it a NEW key is not held (an existing key still updates), counted in <see cref="Dropped"/>.
    /// </summary>
    private sealed class DeferredWrites(int capacity)
    {
        private struct VariableEntry
        {
            public bool HasValue;
            public object? Value;
            public uint Status;
            public DateTime TimestampUtc;
        }

        private readonly Dictionary<(AddressSpaceRealm, string), VariableEntry> _variables = new();
        private readonly Dictionary<(AddressSpaceRealm, string), AlarmStateUpdate> _alarms = new();
        private readonly Dictionary<(AddressSpaceRealm, string), AlarmQualityUpdate> _qualities = new();

        public int Capacity { get; } = capacity;
        public long Dropped { get; private set; }
        public int Count => _variables.Count + _alarms.Count + _qualities.Count;
        public bool IsEmpty => Count == 0;
        public IEnumerable<AlarmStateUpdate> Alarms => _alarms.Values;
        public IEnumerable<AlarmQualityUpdate> AlarmQualities => _qualities.Values;

        public bool Value(in ValueWrite write)
        {
            if (string.IsNullOrEmpty(write.NodeId)) return true;
            var key = (write.Realm, write.NodeId);
            if (!_variables.ContainsKey(key) && !HasRoom()) return false;
            _variables[key] = new VariableEntry
            {
                HasValue = true, Value = write.Value, Status = write.StatusCode, TimestampUtc = write.SourceTimestampUtc,
            };
            return true;
        }

        public bool Status(in StatusWrite write)
        {
            if (string.IsNullOrEmpty(write.NodeId)) return true;
            var key = (write.Realm, write.NodeId);
            if (_variables.TryGetValue(key, out var entry))
            {
                entry.Status = write.StatusCode;
                entry.TimestampUtc = write.SourceTimestampUtc;
                _variables[key] = entry;
                return true;
            }

            if (!HasRoom()) return false;
            _variables[key] = new VariableEntry { Status = write.StatusCode, TimestampUtc = write.SourceTimestampUtc };
            return true;
        }

        /// <summary>A full snapshot also sets the condition's Quality (from its StatusCode), so it SUPERSEDES an
        /// earlier quality-only write for the same id (review N1): replayed state-then-quality, the stale quality
        /// would overwrite the snapshot's. A quality arriving AFTER the state is kept and replays after it.</summary>
        public bool Alarm(AlarmStateUpdate msg)
        {
            var key = (msg.Realm, msg.AlarmNodeId);
            if (!Put(_alarms, key, msg)) return false;
            _qualities.Remove(key);
            return true;
        }

        public bool AlarmQuality(AlarmQualityUpdate msg) => Put(_qualities, (msg.Realm, msg.AlarmNodeId), msg);

        public (List<ValueWrite> Values, List<StatusWrite> Statuses) VariableWrites(Func<(AddressSpaceRealm, string), bool> include)
        {
            var values = new List<ValueWrite>();
            var statuses = new List<StatusWrite>();
            foreach (var ((realm, id), e) in _variables)
            {
                if (!include((realm, id))) continue;
                if (e.HasValue) values.Add(new ValueWrite(id, e.Value, e.Status, e.TimestampUtc, realm));
                else statuses.Add(new StatusWrite(id, e.Status, e.TimestampUtc, realm));
            }

            return (values, statuses);
        }

        public void Clear()
        {
            _variables.Clear();
            _alarms.Clear();
            _qualities.Clear();
            Dropped = 0;
        }

        private bool Put<T>(Dictionary<(AddressSpaceRealm, string), T> map, (AddressSpaceRealm, string) key, T value)
        {
            if (string.IsNullOrEmpty(key.Item2)) return true;
            if (!map.ContainsKey(key) && !HasRoom()) return false;
            map[key] = value;
            return true;
        }

        private bool HasRoom()
        {
            if (Count < Capacity) return true;
            Dropped++;
            return false;
        }
    }

    /// <summary>
    ///     Closes an apply on EVERY exit — early returns and a throwing pass included: the node manager reports
    ///     here the native conditions a rebuild carried over (after the notifier wiring the passes did) and
    ///     discards any captured state no materialise claimed. Skipping it on a throw would leave that state for a
    ///     later surgical re-add of the same id to resurrect. No-op when nothing is pending; never throws. The
    ///     outcome goes out only after the apply (including its completion step) is done.
    /// </summary>
    private void FinishRebuild(RebuildAddressSpace msg, IActorRef requester, (bool Degraded, string? Reason) outcome)
    {
        try { _sink.CompleteAddressSpaceApply(); }
        catch (Exception ex)
        {
            _log.Error(ex, "OpcUaPublish: sink.CompleteAddressSpaceApply threw (correlation={Correlation})", msg.Correlation);
        }

        // Land what arrived while the plan was in flight on the nodes the apply just created — before the answer,
        // so "applied" means the deferred first values are on the address space too (review fix C1).
        ReplayDeferredWrites(_replayScope);
        _replayScope = ReplayScope.Nothing;

        if (msg.ReportOutcome)
            requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, outcome.Degraded, outcome.Reason));
    }

    /// <summary>The actor-thread half of a rebuild: install the artifact's authority, apply the plan through the
    /// sink passes, announce. Returns whether the apply was degraded, and why.</summary>
    private (bool Degraded, string? Reason) ApplyPlan(PlanReady ready)
    {
        var previous = _lastApplied;
        var recording = _applier!.BeginRecording();
        try
        {
            var result = ApplyPlanCore(ready, out var rebuilt);
            // What this apply created or reset (see _replayScope). A throwing pass still scopes to what it did.
            var conditions = AddedConditions(previous, ready.Composition);
            if (rebuilt)
            {
                // Review N3: a full rebuild recreates every SCRIPTED condition fresh — the node manager carries only
                // NATIVE conditions' state over (OtOpcUaNodeManager.RebuildAddressSpace). The engine's re-assert
                // (or any transition) handled in the window landed on the discarded instance, so every kept scripted
                // condition must be replayed onto its new one. Native kept conditions stay "added only".
                foreach (var a in ready.Composition.EquipmentScriptedAlarms)
                    if (a.Enabled) conditions.Add((AddressSpaceRealm.Uns, a.ScriptedAlarmId));
            }

            _replayScope = new ReplayScope(rebuilt, recording.Ensured, recording.Reshaped, conditions);
            return result;
        }
        finally
        {
            _applier.EndRecording();
        }
    }

    /// <summary>The condition ids in <paramref name="next"/> that <paramref name="previous"/> did not materialise:
    /// enabled scripted alarms (Uns, by ScriptedAlarmId) and native alarms (Raw, at their RawPath).</summary>
    private static HashSet<(AddressSpaceRealm, string)> AddedConditions(
        AddressSpaceComposition previous, AddressSpaceComposition next)
    {
        var added = ConditionIds(next);
        added.ExceptWith(ConditionIds(previous));
        return added;

        static HashSet<(AddressSpaceRealm, string)> ConditionIds(AddressSpaceComposition c)
        {
            var ids = new HashSet<(AddressSpaceRealm, string)>();
            foreach (var a in c.EquipmentScriptedAlarms)
                if (a.Enabled) ids.Add((AddressSpaceRealm.Uns, a.ScriptedAlarmId));
            foreach (var t in c.RawTags)
                if (t.Alarm is not null) ids.Add((AddressSpaceRealm.Raw, t.NodeId));
            return ids;
        }
    }

    private (bool Degraded, string? Reason) ApplyPlanCore(PlanReady ready, out bool rebuilt)
    {
        rebuilt = false;
        var msg = ready.Request;
        using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan();
        span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString());

        try
        {
            foreach (var inconsistency in ready.Inconsistencies)
                _log.Warning("OpcUaPublish {Node}: cross-cluster binding — {Message}", _localNode, inconsistency);

            // install this artifact's API keys + ACL grants BEFORE the address-space work
            // below, and deliberately ahead of the `plan.IsEmpty` early-return: an artifact that revokes a
            // key or edits a grant without touching a single node produces an EMPTY address-space plan, so
            // feeding this after that return would leave a revoked key authenticating until some unrelated
            // structural change happened to redeploy. Authorization and topology change on their own
            // schedules.
            if (ready.Authority is not null) FeedDataPlaneAuthority(ready.Authority, ready.Artifact);

            var composition = ready.Composition;
            var plan = ready.Plan;

            if (plan.IsEmpty)
            {
                if (_lastAppliedDegraded)
                {
                    // The composition is unchanged but the last apply of it was a degraded
                    // rebuild: re-ensure every node (each Ensure* no-ops on a node that stands) so the
                    // re-deploy actually repairs what that rebuild could not place.
                    var repairFailed = _applier!.MaterialiseHierarchy(composition)
                        + MaterialiseNodes(plan, composition, fullComposition: true);
                    if (repairFailed > 0)
                    {
                        OtOpcUaTelemetry.OpcUaApplyFailed.Add(1, new KeyValuePair<string, object?>("kind", "nodes"));
                        _log.Error(
                            "OpcUaPublish: re-apply of a degraded composition is STILL degraded (correlation={Correlation}, failedNodes={FailedNodes})",
                            msg.Correlation, repairFailed);
                        return (true, $"re-apply of a previously degraded address space is still degraded (failedNodes={repairFailed})");
                    }

                    _lastAppliedDegraded = false;
                    _log.Info("OpcUaPublish: re-apply repaired the previously degraded address space (correlation={Correlation})",
                        msg.Correlation);
                    return (false, null);
                }

                _log.Debug("OpcUaPublish: rebuild requested but plan is empty (correlation={Correlation})",
                    msg.Correlation);
                return (false, null);
            }

            var outcome = _applier!.Apply(plan);
            rebuilt = outcome.RebuildCalled;

            // Sum swallowed per-node materialise failures across every pass together with Apply's own
            // removal-pass tally (archreview 01/S-1). A degraded apply used to vanish into per-node
            // Warnings + an optimistic Info line; now it surfaces at Error + a dedicated meter.
            var failedNodes = outcome.FailedNodes;
            // The UNS Area/Line/Equipment folder hierarchy is materialised in FULL on every apply, rebuild or
            // not: it is O(areas + lines + equipment) — trivial next to the tag counts the delta passes below
            // exist to bound — and every node the delta passes add needs its ancestor folders to already
            // exist, so narrowing it would mean re-deriving each add's ancestor closure (strictly more work).
            failedNodes += _applier.MaterialiseHierarchy(composition);
            failedNodes += MaterialiseNodes(plan, composition, fullComposition: outcome.RebuildCalled);

            // 2026-08-15 surgical-deploys review C1 — record the composition as applied ONLY when the apply was
            // clean, or when a rebuild ran. Two reasons this sits HERE (after the materialise passes) rather
            // than immediately after Apply:
            //
            //  • A pass that THROWS outright leaves _lastApplied at the previous composition (the outer catch
            //    below takes over), so the next deploy re-diffs against what is actually materialised.
            //  • A DEGRADED non-rebuild apply (failedNodes > 0, no rebuild) must not advance either. Before the
            //    delta materialisation the full passes re-ensured every node on every apply, which silently
            //    repaired any prior swallowed sink fault — an accidental self-healing mechanism. The delta path
            //    removed it: a node whose EnsureVariable failed is inside _lastApplied, never appears in a diff
            //    set again, and EnsureVariable early-returns on the ones that DID land — so the failed node is
            //    lost until some unrelated change happens to force a rebuild. Holding the composition back is
            //    the replacement: the next deploy re-diffs the same composition and re-ensures what failed.
            //
            // What the re-diff costs, stated honestly: it also re-emits the previous apply's REMOVES. A
            // re-remove of an already-removed node returns false from the surgical sink, which trips the
            // applier's one-way ratchet into a full rebuild — the correct resync for an apply that has already
            // been logged at Error as degraded, and a bounded one (it happens once, then the address space is
            // known-good again).
            //
            // A rebuild DOES advance even with failures: it already re-materialised the whole composition, and
            // the ratchet is the resync path for anything it could not place. Holding it back would make every
            // subsequent deploy re-diff the entire address space as added.
            if (outcome.RebuildCalled || failedNodes == 0)
            {
                _lastApplied = composition;
                _hasAppliedComposition = true;
                // Advanced despite failures only on a rebuild; remember it so an identical
                // re-deploy repairs instead of short-circuiting clean on the empty plan.
                _lastAppliedDegraded = outcome.RebuildFailed || failedNodes > 0;
            }
            else
            {
                _log.Error(
                    "OpcUaPublish: degraded non-rebuild apply (correlation={Correlation}, failedNodes={FailedNodes}) — NOT recording this composition as applied, so the next deploy re-diffs it and repairs the missing node(s)",
                    msg.Correlation, failedNodes);
            }

            // On a NON-rebuild apply (PureAdd / AttributeOnly), the Materialise passes above
            // just created exactly the added nodes WITHOUT tearing anything down, so announce them with a
            // Part 3 GeneralModelChangeEvent(NodeAdded) per affected parent — model-aware clients then
            // re-browse and pick up the new nodes while every existing MonitoredItem stays alive. Ordering
            // is correct by construction: Apply returned before the passes ran, and this announce runs after
            // them, so the nodes exist when clients re-browse. After a full rebuild the announcement is moot
            // (subscriptions are dead anyway) — the guard skips it. AnnounceAddedNodes is Safe-wrapped
            // internally, so a faulting announce can never break the deploy.
            //
            // The announce is realm-aware: UNS parents in the Uns realm, and the parents of
            // added RAW containers/tags in the Raw realm. It matters because of the delta work — before it, a
            // plan carrying raw adds went through a full RebuildAddressSpace, so every client re-browsed
            // anyway; now those adds are materialised surgically while existing subscriptions stay alive, and
            // an unannounced add would stay invisible until a client happened to browse again.
            if (!outcome.RebuildCalled) _applier.AnnounceAddedNodes(plan);

            OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "rebuild"));
            if (outcome.RebuildFailed || failedNodes > 0)
            {
                OtOpcUaTelemetry.OpcUaApplyFailed.Add(1,
                    new KeyValuePair<string, object?>("kind", outcome.RebuildFailed ? "rebuild" : "nodes"));
                _log.Error(
                    "OpcUaPublish: address-space apply DEGRADED (correlation={Correlation}, rebuildFailed={RebuildFailed}, failedNodes={FailedNodes}, added={Added}, removed={Removed}, changed={Changed}) — the running address space may be stale or partial",
                    msg.Correlation, outcome.RebuildFailed, failedNodes,
                    outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes);
                return (true, outcome.RebuildFailed
                    ? $"address-space rebuild failed (failedNodes={failedNodes})"
                    : $"address-space apply degraded (failedNodes={failedNodes}); not recorded as applied, the next deploy of this revision re-diffs and repairs it");
            }

            _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})",
                msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled);
            return (false, null);
        }
        catch (Exception ex)
        {
            _log.Error(ex, "OpcUaPublish: rebuild pipeline threw (correlation={Correlation})", msg.Correlation);
            return (true, $"address-space apply threw: {ex.Message}");
        }
    }

    /// <summary>
    ///     Run the four node-materialising applier passes — Raw subtree, UNS references, scripted alarms,
    ///     equipment VirtualTags — either over the FULL composition (after a rebuild) or over
    ///     just this plan's delta (surgical deploys). <c>MaterialiseHierarchy</c> is NOT one of them:
    ///     the caller always runs it in full, because parents must exist before any add.
    /// </summary>
    /// <remarks>
    ///     <para><b>Why the delta path exists.</b> The passes are idempotent, but idempotent is not free.
    ///     <c>EnsureFolder</c> / <c>EnsureVariable</c> early-return on a <c>ConcurrentDictionary</c> hit
    ///     BEFORE taking the node manager's lock, so an already-standing node costs a dictionary probe rather
    ///     than a lock acquisition — but <c>MaterialiseAlarmCondition</c> DOES take the lock every time, and
    ///     the probes are still O(address space) per apply plus the composition walk that produces them.
    ///     Running the passes over the whole composition made the cost of ANY apply proportional to the size
    ///     of the served address space rather than to the size of the change: adding one tag to a 50k-node
    ///     server was ~50k sink calls on the actor's pinned publish thread, with every client's publish
    ///     blocked behind it. The delta path makes it O(change).</para>
    ///     <para><b>What the full pass was silently doing, and its replacement.</b> Re-ensuring every node on
    ///     every apply also REPAIRED any node a previous apply had swallowed a sink fault on — accidental
    ///     self-healing nobody designed. The caller's degraded-apply no-advance rule (a non-rebuild apply
    ///     reporting <c>failedNodes &gt; 0</c> does not record its composition in <c>_lastApplied</c>, so the
    ///     next deploy re-diffs and re-ensures it) is the deliberate replacement for it.</para>
    ///     <para><b>Why <see cref="AddressSpaceApplyOutcome.RebuildCalled"/> is the branch key</b> rather than
    ///     the classifier's kind: a surgical apply that hits a <c>false</c>/throw from the surgical sink falls
    ///     back to a full <c>RebuildAddressSpace()</c> INSIDE <c>Apply</c>, which wipes every node. That plan
    ///     is still classified non-Rebuild, so branching on the kind would re-materialise only the delta onto
    ///     a wiped address space. The outcome flag reports what actually happened.</para>
    ///     <para><b>Why changed UNS references are in the subset.</b> <c>Apply</c> repairs a
    ///     <c>UnsReferenceDelta</c> by REMOVING the previous node and leaving the recreate to this pass, so
    ///     the subset must carry the <i>current</i> side of every changed reference as well as the added ones
    ///     — otherwise a re-pointed, renamed, or backing-edited UNS node is removed and never comes back.</para>
    ///     <para><b>Accepted narrowing.</b> The plan diffs the THIN <c>ScriptedAlarmPlans</c> projection, so
    ///     the added-alarm subset is the rich <c>EquipmentScriptedAlarms</c> rows whose id the plan reports as
    ///     added. An alarm already present in both compositions whose predicate SCRIPT row only now arrived
    ///     (making it materialisable for the first time) is not in that set — but neither is it in any diff
    ///     set at all, so on its own it already produced an empty plan and materialised nothing. The delta
    ///     path extends that pre-existing blindness to the case where it rides along with another change.</para>
    /// </remarks>
    /// <param name="plan">The plan just applied — the source of the delta subsets.</param>
    /// <param name="composition">The full composition, always passed as the passes' lookup context.</param>
    /// <param name="fullComposition">When true, materialise the whole composition (post-rebuild).</param>
    /// <returns>The summed count of swallowed per-node sink failures across the passes that ran.</returns>
    private int MaterialiseNodes(AddressSpacePlan plan, AddressSpaceComposition composition, bool fullComposition)
    {
        var applier = _applier!;
        var failedNodes = 0;

        // v3 Batch 4 — the Raw device subtree (ns=Raw): containers (Folder→Driver→Device→TagGroup) then
        // tag Variables keyed by RawPath. Materialised BEFORE the UNS references so each UNS Organizes→Raw
        // edge finds its raw target. The driver binds live values to these raw NodeIds.
        if (fullComposition)
        {
            failedNodes += applier.MaterialiseRawSubtree(composition);
        }
        else if (plan.AddedRawContainers.Count > 0 || plan.AddedRawTags.Count > 0)
        {
            failedNodes += applier.MaterialiseRawSubtree(composition, plan.AddedRawContainers, plan.AddedRawTags);
        }

        // v3 Batch 4 — the UNS reference Variables (ns=UNS): each projects a raw tag under its equipment
        // folder (created by MaterialiseHierarchy) with an Organizes→Raw edge; values fan out from the raw
        // node. Runs AFTER both the equipment folders AND the raw subtree exist.
        if (fullComposition)
        {
            failedNodes += applier.MaterialiseUnsReferences(composition);
        }
        else
        {
            var unsReferences = DeltaUnsReferences(plan);
            if (unsReferences.Count > 0) failedNodes += applier.MaterialiseUnsReferences(composition, unsReferences);
        }

        // T14 — scripted alarms get their own pass right after the hierarchy so the equipment
        // folders they parent under already exist. Materialises real Part 9 AlarmConditionState
        // nodes (keyed by ScriptedAlarmId so AlarmStateUpdate writes target them); disabled
        // alarms are skipped.
        if (fullComposition)
        {
            failedNodes += applier.MaterialiseScriptedAlarms(composition);
        }
        else
        {
            var addedAlarms = DeltaScriptedAlarms(plan, composition);
            if (addedAlarms.Count > 0) failedNodes += applier.MaterialiseScriptedAlarms(composition, addedAlarms);
        }

        // Equipment-namespace VirtualTags get their own pass right after the scripted alarms:
        // ensures each computed signal's Variable (and any FolderPath sub-folder) exists under its
        // equipment folder with a folder-scoped NodeId. VirtualTagHostActor.OnResult pushes live
        // values once the first dependency update arrives; until then variables show BadWaitingForInitialData.
        if (fullComposition)
        {
            failedNodes += applier.MaterialiseEquipmentVirtualTags(composition);
        }
        else if (plan.AddedEquipmentVirtualTags.Count > 0)
        {
            failedNodes += applier.MaterialiseEquipmentVirtualTags(composition, plan.AddedEquipmentVirtualTags);
        }

        return failedNodes;
    }

    /// <summary>The UNS reference variables a non-rebuild apply must materialise: the plan's ADDED references
    /// plus the CURRENT side of every CHANGED one (whose previous node the applier already removed in place).
    /// </summary>
    /// <param name="plan">The plan just applied.</param>
    /// <returns>The references to materialise (possibly empty).</returns>
    private static IReadOnlyList<UnsReferenceVariable> DeltaUnsReferences(AddressSpacePlan plan)
    {
        if (plan.ChangedUnsReferenceVariables.Count == 0) return plan.AddedUnsReferenceVariables;
        var list = new List<UnsReferenceVariable>(
            plan.AddedUnsReferenceVariables.Count + plan.ChangedUnsReferenceVariables.Count);
        list.AddRange(plan.AddedUnsReferenceVariables);
        foreach (var d in plan.ChangedUnsReferenceVariables) list.Add(d.Current);
        return list;
    }

    /// <summary>The scripted-alarm host plans a non-rebuild apply must materialise: the rich
    /// <c>EquipmentScriptedAlarms</c> rows whose <c>ScriptedAlarmId</c> the plan reports as ADDED (the plan
    /// diffs only the thin projection — see the narrowing note on <see cref="MaterialiseNodes"/>).</summary>
    /// <param name="plan">The plan just applied.</param>
    /// <param name="composition">The full composition carrying the rich alarm plans.</param>
    /// <returns>The alarms to materialise (possibly empty).</returns>
    private static IReadOnlyList<EquipmentScriptedAlarmPlan> DeltaScriptedAlarms(
        AddressSpacePlan plan, AddressSpaceComposition composition)
    {
        if (plan.AddedAlarms.Count == 0 || composition.EquipmentScriptedAlarms.Count == 0)
            return Array.Empty<EquipmentScriptedAlarmPlan>();

        var addedIds = new HashSet<string>(plan.AddedAlarms.Select(a => a.ScriptedAlarmId), StringComparer.Ordinal);
        return composition.EquipmentScriptedAlarms.Where(a => addedIds.Contains(a.ScriptedAlarmId)).ToList();
    }

    /// <summary>
    ///     Decode the artifact's data-plane authorization inputs (API keys + this cluster's ACL grants)
    ///     and install them for OPC UA session activation. No-op when nothing consumes
    ///     them.
    /// </summary>
    /// <remarks>
    ///     Wrapped so a decode failure can never break a deploy — the same discipline as the historised-ref
    ///     feed. The cost of that choice is explicit: the node keeps serving the PREVIOUS key set, which is
    ///     the safe direction for availability but means a revocation carried by an artifact this node
    ///     could not decode has not taken effect. It is logged at Error for exactly that reason.
    /// </remarks>
    /// <param name="projection">The keys + grants decoded (off-thread) from the rebuild's single parse of the artifact.</param>
    /// <param name="artifactBytes">The RAW artifact bytes — hashed for the revision id, which must be
    /// SHA-256 of the exact blob (see below); never re-parsed here.</param>
    private void FeedDataPlaneAuthority(DataPlaneAuthorityProjection projection, byte[] artifactBytes)
    {
        if (_dataPlaneAuthority is null) return;

        try
        {
            // The revision hash is SHA-256 of the artifact bytes — the same derivation ConfigComposer uses,
            // which is what lets a node recognise a re-applied revision without the hash riding the wire.
            // Hashed off the RAW bytes for exactly that reason: a re-serialisation of the parsed document
            // would be a different byte sequence and would not match the deployment's RevisionHash.
            var revision = Convert.ToHexStringLower(System.Security.Cryptography.SHA256.HashData(artifactBytes));
            var snapshot = _dataPlaneAuthority.Apply(
                projection.ClusterId, revision, projection.Keys, projection.Grants, projection.ScopePaths);

            _log.Info(
                "OpcUaPublish: data-plane authority generation {Generation} installed (cluster={Cluster}, revision={Revision}, keys={Keys}, grants={Grants}, resolvedScopes={ResolvedScopes})",
                snapshot.GenerationId, projection.ClusterId ?? "(single-cluster)", revision[..12],
                projection.Keys.Count, projection.Grants.Count, projection.ScopePaths.Count);

            // Reported BY the trie build, not re-derived here. The previous version of this
            // log counted unresolvable ScopeIds itself and so agreed with the builder on exactly one of its
            // four drop reasons — a row with a blank ScopeId, a retired scope kind, or a scope path that
            // does not start at a realm all vanished with the log still reading zero. A dropped row is an
            // authored rule that can never match, and this is the operator's only warning of it.
            if (snapshot.DroppedGrants.Count > 0)
            {
                _log.Warning(
                    "OpcUaPublish: {Dropped} ACL rule(s) in generation {Generation} were DROPPED and will never match — {Detail}",
                    snapshot.DroppedGrants.Count, snapshot.GenerationId,
                    string.Join("; ", snapshot.DroppedGrants
                        .Select(d => $"acl {d.NodeAclId} ({d.ScopeKind} '{d.ScopeId}') {d.Reason}")));
            }

            if (string.IsNullOrWhiteSpace(projection.ClusterId) && projection.Grants.Count > 0)
            {
                // No cluster identity means no trie was built at all, so nothing can be granted here. Say
                // so explicitly: with enforcement on, this node denies every write, alarm command and
                // HistoryRead, and the grant count logged above would otherwise suggest the opposite.
                _log.Warning(
                    "OpcUaPublish: this node has no cluster identity, so NO permission trie was built for its {Grants} grant(s) — with Security:Acl:Enabled every scoped decision here will be denied",
                    projection.Grants.Count);
            }
        }
        catch (Exception ex)
        {
            _log.Error(ex,
                "OpcUaPublish: failed to install data-plane authority from the applied artifact — the node keeps the PREVIOUS keys and grants, so any revocation in this deployment has NOT taken effect");
        }
    }

    private void HandleServiceLevelChanged(ServiceLevelChanged msg)
    {
        // Always publish the FIRST computed level, even if it equals the byte-default 0. Otherwise a
        // node starting Detached/role-less (first level = 0) would be dedup'd away, leaving the SDK's
        // built-in default (255 = full service) standing — a degraded node wrongly advertising 255.
        if (_publishedAtLeastOnce && msg.ServiceLevel == _lastServiceLevel) return;
        try
        {
            _serviceLevel.Publish(msg.ServiceLevel);
            _lastServiceLevel = msg.ServiceLevel;
            _publishedAtLeastOnce = true;
            OtOpcUaTelemetry.ServiceLevelChange.Add(1,
                new KeyValuePair<string, object?>("level", msg.ServiceLevel));
            _log.Debug("OpcUaPublish: ServiceLevel={Level}", msg.ServiceLevel);
        }
        catch (Exception ex)
        {
            // A failed write may be partially applied; even a previous successful value must retry.
            _publishedAtLeastOnce = false;
            _log.Warning(ex, "OpcUaPublish: ServiceLevel publisher threw at level {Level}", msg.ServiceLevel);
        }
    }

    /// <summary>Caches the latest redundancy snapshot and recomputes the local ServiceLevel.
    /// The actual byte is produced by <see cref="RecomputeServiceLevel"/> — either via the
    /// health-aware <see cref="ServiceLevelCalculator"/> (once a DB-health probe + sample are
    /// wired) or via the legacy role-only seam (back-compat / bootstrap).</summary>
    private void HandleRedundancyStateChanged(RedundancyStateChanged msg)
    {
        _lastSnapshot = msg;
        RecomputeServiceLevel();
    }

    /// <summary>Caches the latest DB-health sample and recomputes the local ServiceLevel. The
    /// probe pushes these (or the actor Asks for them); either way the freshest sample feeds the
    /// calculator's <c>DbReachable</c>/<c>Stale</c> inputs.</summary>
    private void HandleDbHealthStatus(DbHealthProbeActor.DbHealthStatus msg)
    {
        _lastDbHealth = msg;
        RecomputeServiceLevel();
    }

    /// <summary>Periodic self-tick: recomputes freshness directly when no DB probe exists; otherwise
    /// Asks the local <see cref="DbHealthProbeActor"/> for its cached
    /// status and pipes the reply back to <see cref="Self"/>, where <see cref="HandleDbHealthStatus"/>
    /// caches it + recomputes. A hung/late probe (Ask timeout) pipes a <c>Reachable=false</c> status so
    /// the node FAIL-SAFE-DEMOTES rather than freezing the last-known-good. The continuation only
    /// CONSTRUCTS a record off the actor thread — the actual state mutation happens on the actor thread
    /// when the piped <see cref="DbHealthProbeActor.DbHealthStatus"/> is received, so no actor field is
    /// touched here.</summary>
    private void OnHealthTick()
    {
        if (_dbHealthProbe is null)
        {
            RecomputeServiceLevel();
            return;
        }

        _dbHealthProbe.Ask<DbHealthProbeActor.DbHealthStatus>(
                DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(1))
            .ContinueWith(t => t.IsCompletedSuccessfully
                ? t.Result
                : new DbHealthProbeActor.DbHealthStatus(false, _utcNow(), "db-health ask timeout"))
            .PipeTo(Self);
    }

    /// <summary>Records a peer's OPC UA probe verdict about THIS node and recomputes the local
    /// ServiceLevel. The probe's <see cref="PeerOpcUaProbeActor.OpcUaProbeResult.NodeId"/> is the
    /// target that was probed, so a result whose <c>NodeId</c> is not this node is about a peer and
    /// is ignored. A matching result is stamped with the receive time so <see cref="OpcUaProbeOk"/>
    /// can debounce stale verdicts.</summary>
    private void HandlePeerProbe(PeerOpcUaProbeActor.OpcUaProbeResult r)
    {
        // The result targets the probed node. If it isn't me, it's about a peer — ignore it.
        if (_localNode is null || r.NodeId != _localNode.Value) return;
        _probeAboutMe = (r.Ok, _utcNow());
        RecomputeServiceLevel();
    }

    /// <summary>The OPC UA self-probe input for the calculator: "did a peer recently observe MY OPC UA
    /// endpoint as reachable?" Returns <c>true</c> (benefit of the doubt) when no peer verdict has
    /// arrived yet (single-node / no peer) or when the latest verdict is older than
    /// <see cref="_probeFreshnessWindow"/> (the peer went away — don't penalise this node for that).
    /// Only an actively-observed, RECENT <c>Ok==false</c> demotes.</summary>
    private bool OpcUaProbeOk()
    {
        if (_probeAboutMe is not { } verdict) return true;
        if (_utcNow() - verdict.At > _probeFreshnessWindow) return true;
        return verdict.Ok;
    }

    /// <summary>The OPC UA endpoint input for the calculator (lmxopcua#762): <c>false</c> only when this node's own
    /// SDK start is recorded as <see cref="OpcUaEndpointStatus.Failed"/>. <see cref="OpcUaEndpointStatus.NotStarted"/>
    /// and <see cref="OpcUaEndpointStatus.Stopped"/> are boot and shutdown windows, not faults, and a node with no
    /// endpoint fact at all (a harness) is treated as listening. Read on every recompute — the periodic health tick
    /// and every redundancy snapshot (heartbeated every 10 s) — so a change is picked up without a notification.
    /// It is a LOCAL fact on purpose: the peer TCP probe behind <see cref="OpcUaProbeOk"/> cannot see a failed start
    /// when another process holds the port, because its connect then succeeds against that process.</summary>
    private bool EndpointListening() => _endpointState?.Status != OpcUaEndpointStatus.Failed;

    /// <summary>
    ///     Computes the local OPC UA ServiceLevel and routes it through <see cref="ServiceLevelChanged"/>
    ///     (the dedup/publish/metric handler). The full <see cref="ServiceLevelCalculator"/> path is
    ///     used once a DB-health probe is wired AND a sample has arrived; until then (and when no probe
    ///     is supplied at all) a legacy role-only seam keeps the historical "primary-leader → 240,
    ///     secondary → 100, detached → 0" behaviour. The calculator does not model Detached, so a
    ///     detached local node is guarded to 0 before either path runs.
    /// </summary>
    private void RecomputeServiceLevel()
    {
        if (_localNode is null || _lastSnapshot is null) return;

        var entry = _lastSnapshot.Nodes.FirstOrDefault(n => n.NodeId == _localNode.Value);

        // The calculator does NOT model Detached — a healthy detached node would wrongly compute
        // 240, so guard it (and the missing-entry case) to 0 here.
        if (entry is null || entry.Role == RedundancyRole.Detached)
        {
            Self.Tell(new ServiceLevelChanged(0));
            return;
        }

        // Per-cluster mesh Phase 4: on a driver-only (DB-less) node there is no ConfigDb to probe —
        // LocalDb IS the config store and is in-process, so it can never be "unreachable". Retire the
        // DB-reachability input by feeding DbReachable=true CONSTANT and deriving staleness from the
        // snapshot age ALONE (no DB-health term). This is the survive-alone posture and it is
        // client-visible: a healthy DB-less node must publish 240 (240+10=250 as Primary) even with
        // central down, so clients keep steering to it — the old DbReachable=false would have computed
        // the (false,true,false) → 0 arm and driven clients AWAY from a node serving perfectly.
        // Staleness ("running behind on config") still demotes to 200. The Detached guard above still
        // wins, so a DB-less Detached node stays 0.
        if (_dbLess)
        {
            var utcNow = _utcNow();
            var dbLessInputs = new NodeHealthInputs(
                MemberState: SafeSelfStatus(),
                DbReachable: true,
                OpcUaProbeOk: OpcUaProbeOk(),
                Stale: (utcNow - entry.AsOfUtc) > _staleWindow,
                IsDriverPrimary: entry.IsDriverPrimary,
                EndpointListening: EndpointListening());
            Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(dbLessInputs)));
            return;
        }

        // Legacy / back-compat seam: with no DB-health probe wired (or before the first sample
        // arrives) fall back to the old role-only switch. This preserves historical behaviour and
        // is the bootstrap value until the first DbHealthStatus lands.
        if (_dbHealthProbe is null || _lastDbHealth is null)
        {
            // The role-only seam has no health inputs, but a failed endpoint still wins (lmxopcua#762): the
            // legacy switch would otherwise hand a node that serves nothing 240 as Primary.
            Self.Tell(new ServiceLevelChanged(EndpointListening() ? LegacyRoleOnly(entry) : (byte)0));
            return;
        }

        var now = _utcNow();
        var inputs = new NodeHealthInputs(
            MemberState: SafeSelfStatus(),
            DbReachable: _lastDbHealth.Reachable,
            OpcUaProbeOk: OpcUaProbeOk(),
            Stale: !_lastDbHealth.Reachable
                   || (now - _lastDbHealth.AsOfUtc) > _staleWindow
                   || (now - entry.AsOfUtc) > _staleWindow,
            IsDriverPrimary: entry.IsDriverPrimary,
            EndpointListening: EndpointListening());

        Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(inputs)));
    }

    /// <summary>The legacy role-only ServiceLevel switch (primary-leader → 240, primary → 200,
    /// secondary → 100, _ → 0). Preserved as the back-compat / bootstrap seam.</summary>
    private static byte LegacyRoleOnly(NodeRedundancyState entry) => entry.Role switch
    {
        RedundancyRole.Primary when entry.IsDriverPrimary => 240,
        RedundancyRole.Primary => 200,
        RedundancyRole.Secondary => 100,
        _ => 0,
    };

    /// <summary>Reads this node's cluster <see cref="MemberStatus"/>, returning
    /// <see cref="MemberStatus.Removed"/> if the cluster is unavailable (so the calculator treats it
    /// as untrusted → 0 rather than throwing).</summary>
    private MemberStatus SafeSelfStatus()
    {
        try
        {
            return _cluster.SelfMember.Status;
        }
        catch (Exception ex)
        {
            _log.Debug(ex, "OpcUaPublish: SelfMember status unavailable; treating as Removed (ServiceLevel→0)");
            return MemberStatus.Removed;
        }
    }
}