ScadaBridge/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs

using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.Singleton;
using Akka.Configuration;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.ClusterInfrastructure;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
using ZB.MOM.WW.ScadaBridge.Host.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming;
using ZB.MOM.WW.ScadaBridge.StoreAndForward;

namespace ZB.MOM.WW.ScadaBridge.Host.Actors;

/// <summary>
/// Hosted service that manages the Akka.NET actor system lifecycle.
/// Creates the actor system on start, registers actors, and triggers
/// CoordinatedShutdown on stop.
///
/// Transport heartbeat is explicitly configured in HOCON from CommunicationOptions.
/// </summary>
public class AkkaHostedService : IHostedService
{
    private readonly IServiceProvider _serviceProvider;
    private readonly NodeOptions _nodeOptions;
    private readonly ClusterOptions _clusterOptions;
    private readonly CommunicationOptions _communicationOptions;
    private readonly ILogger<AkkaHostedService> _logger;
    private readonly IHostApplicationLifetime? _appLifetime;
    private ActorSystem? _actorSystem;

    /// <summary>
    /// Flips to <c>true</c> the moment <see cref="StopAsync"/> begins, so the
    /// <see cref="ActorSystem.WhenTerminated"/> watchdog can tell an
    /// intentional host shutdown apart from an SBR self-down that terminates the
    /// system out from under a still-running process. <c>volatile</c>: written on
    /// the shutdown thread, read on the Akka termination-callback thread.
    /// </summary>
    private volatile bool _stopRequested;

    /// <summary>
    /// Guards the one-time creation of <see cref="_actorSystem"/> in
    /// <see cref="GetOrCreateActorSystem"/> so <see cref="StartAsync"/> and a concurrent
    /// health-probe resolution of the DI <see cref="ActorSystem"/> singleton race to create
    /// it exactly once.
    /// </summary>
    private readonly object _actorSystemLock = new();
    /// <summary>
    /// Auxiliary IDisposables (e.g. the SiteAuditTelemetryStalledTracker)
    /// that this hosted service constructs at start time and must tear down
    /// on shutdown — they don't fit the ActorSystem lifecycle but share its
    /// process scope.
    /// </summary>
    private readonly List<IDisposable> _trackedDisposables = new();

    /// <summary>
    /// Sentinel that flips to <c>true</c> the
    /// first time a Notification-category S&amp;F delivery handler is registered
    /// on this hosted service instance. <see cref="StoreAndForwardService.RegisterDeliveryHandler"/>
    /// is last-write-wins on category, so a future code change that introduces
    /// a second registration path (e.g. a role-branch + helper that both call
    /// the registration) would silently overwrite the canonical
    /// <c>NotificationForwarder</c> handler with whatever the loser registers —
    /// a prior fix did exactly this, and was silently superseded
    /// when the central-only redesign moved delivery to <c>NotificationOutbox</c>.
    /// This sentinel makes the duplicate noisy at startup so a maintainer
    /// re-introducing the second path sees it immediately.
    /// </summary>
    private bool _notificationDeliveryHandlerRegistered;

    /// <summary>
    /// Initializes a new instance of the <see cref="AkkaHostedService"/> class.
    /// </summary>
    /// <param name="serviceProvider">The service provider for accessing dependencies.</param>
    /// <param name="nodeOptions">The node configuration options.</param>
    /// <param name="clusterOptions">The cluster configuration options.</param>
    /// <param name="communicationOptions">The communication configuration options.</param>
    /// <param name="logger">The logger instance.</param>
    /// <param name="appLifetime">The host application lifetime; optional so existing construction sites (tests, DI variants) keep working without it.</param>
    public AkkaHostedService(
        IServiceProvider serviceProvider,
        IOptions<NodeOptions> nodeOptions,
        IOptions<ClusterOptions> clusterOptions,
        IOptions<CommunicationOptions> communicationOptions,
        ILogger<AkkaHostedService> logger,
        IHostApplicationLifetime? appLifetime = null)
    {
        _serviceProvider = serviceProvider;
        _nodeOptions = nodeOptions.Value;
        _clusterOptions = clusterOptions.Value;
        _communicationOptions = communicationOptions.Value;
        _logger = logger;
        // Optional so existing construction sites (tests, DI variants) keep
        // compiling. When absent the watchdog logs but can't stop the process.
        _appLifetime = appLifetime;
    }

    /// <summary>
    /// Gets the actor system once started. Null before StartAsync completes.
    /// </summary>
    public ActorSystem? ActorSystem => _actorSystem;

    /// <summary>
    /// Starts the Akka.NET actor system and registers actors.
    /// </summary>
    /// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
    /// <returns>A task representing the asynchronous operation.</returns>
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        // Create (or reuse) the externally-owned, process-singleton ActorSystem. A
        // health probe may already have created it via the DI singleton bridge
        // (GetOrCreateActorSystem) before this hosted service's StartAsync ran; either way the
        // call yields the one instance and sets _actorSystem. Actor registration below then
        // runs on it.
        var actorSystem = GetOrCreateActorSystem();

        // Register the dead letter monitor actor
        var loggerFactory = _serviceProvider.GetRequiredService<ILoggerFactory>();
        var dlmLogger = loggerFactory.CreateLogger<DeadLetterMonitorActor>();
        var dlmHealthCollector = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.HealthMonitoring.ISiteHealthCollector>();
        actorSystem.ActorOf(
            Props.Create(() => new DeadLetterMonitorActor(dlmLogger, dlmHealthCollector)),
            "dead-letter-monitor");

        // Register role-specific actors
        if (_nodeOptions.Role.Equals("Central", StringComparison.OrdinalIgnoreCase))
        {
            RegisterCentralActors();
        }
        else if (_nodeOptions.Role.Equals("Site", StringComparison.OrdinalIgnoreCase))
        {
            await RegisterSiteActorsAsync(cancellationToken);
        }
    }

    /// <summary>
    /// Returns the process-wide Akka <see cref="ActorSystem"/>, creating it on first call.
    /// Idempotent and thread-safe: both <see cref="StartAsync"/> and the DI bridge that
    /// exposes the system to the shared <c>ZB.MOM.WW.Health.Akka</c> checks call this, and
    /// whichever runs first creates the system exactly once.
    /// </summary>
    /// <remarks>
    /// The <see cref="ActorSystem"/> is an externally-owned process singleton — its
    /// lifecycle is this hosted service's (created here, torn down via
    /// <c>CoordinatedShutdown</c> in <see cref="StopAsync"/>). It MUST be registered in DI as a
    /// <b>singleton resolved through this method</b>, never as a transient/scoped factory:
    /// <see cref="ActorSystem"/> is <see cref="IDisposable"/>, and a transient/scoped factory
    /// hands a fresh disposable to every resolving child scope (e.g. each per-probe
    /// health-check scope), so the container disposes it when that scope ends —
    /// <c>ActorSystem.Dispose()</c> runs <c>CoordinatedShutdown(ActorSystemTerminateReason)</c>
    /// and tears the live cluster node down mid-flight, which is exactly the
    /// "central report pages hang 30s" defect this method fixes. Creating the system here and
    /// exposing it as a singleton keeps child-scope disposal away from it; routing the singleton
    /// through this method (rather than a plain <c>AddSingleton(sp =&gt; ...ActorSystem)</c>
    /// factory) also avoids caching a <c>null</c> if a health probe wins the startup race, since
    /// the first resolve creates the system instead of capturing a not-yet-started reference.
    /// </remarks>
    /// <returns>The single live actor system.</returns>
    public ActorSystem GetOrCreateActorSystem()
    {
        if (_actorSystem is not null)
        {
            return _actorSystem;
        }

        lock (_actorSystemLock)
        {
            if (_actorSystem is not null)
            {
                return _actorSystem;
            }

            // For site nodes, include a site-specific role (e.g., "site-SiteA") alongside the base role
            var roles = BuildRoles();

            // HOCON is assembled in a dedicated builder that quotes/escapes every
            // interpolated value, so a hostname, seed node or strategy containing a quote,
            // backslash or whitespace cannot corrupt the configuration document.
            var hocon = BuildHocon(_nodeOptions, _clusterOptions, roles,
                _communicationOptions.TransportHeartbeatInterval,
                _communicationOptions.TransportFailureThreshold);

            var config = ConfigurationFactory.ParseString(hocon);
            var system = ActorSystem.Create("scadabridge", config);

            _logger.LogInformation(
                "Akka.NET actor system 'scadabridge' started. Role={Role}, Roles={Roles}, Hostname={Hostname}, Port={Port}, " +
                "TransportHeartbeat={TransportHeartbeat}s, TransportFailure={TransportFailure}s",
                _nodeOptions.Role,
                string.Join(", ", roles),
                _nodeOptions.NodeHostname,
                _nodeOptions.RemotingPort,
                _communicationOptions.TransportHeartbeatInterval.TotalSeconds,
                _communicationOptions.TransportFailureThreshold.TotalSeconds);

            // Downed-node recovery watchdog: any downing decision against this node
            // (auto-down by the peer, or an SBR self-down under keep-oldest) plus
            // run-coordinated-shutdown-when-down means a downed node terminates
            // its own ActorSystem. If that happens outside our StopAsync, the Host
            // process must exit so the service supervisor (docker
            // `restart: unless-stopped` / Windows service recovery) restarts it and
            // it rejoins as a fresh incarnation. Without this the node idles forever
            // with a dead actor system (review 01 underdeveloped finding).
            system.WhenTerminated.ContinueWith(_ =>
            {
                if (_stopRequested) return;
                _logger.LogCritical(
                    "ActorSystem terminated outside host shutdown (SBR down / run-coordinated-shutdown-when-down). "
                    + "Stopping the host process so the service supervisor restarts this node as a fresh incarnation.");
                _appLifetime?.StopApplication();
            }, TaskContinuationOptions.ExecuteSynchronously);

            // Publish last so a concurrent reader never observes a half-constructed system.
            _actorSystem = system;
            return _actorSystem;
        }
    }

    /// <summary>
    /// Builds the Akka HOCON configuration document. Every interpolated value is
    /// routed through <see cref="QuoteHocon"/> (string values) so a hostname,
    /// seed-node URI, role or split-brain strategy containing a quote, backslash or
    /// whitespace cannot corrupt the document or be silently misparsed.
    ///
    /// The downing block branches on <see cref="ClusterOptions.SplitBrainResolverStrategy"/>:
    /// <c>auto-down</c> (default; decision 2026-07-21) installs Akka's
    /// <c>AutoDowning</c> provider with <c>auto-down-unreachable-after</c> =
    /// <see cref="ClusterOptions.StableAfter"/> — the leader among the REACHABLE members
    /// downs the unreachable peer, so a crash of either node (oldest included) fails
    /// over to the survivor; the accepted trade is dual-active during a real network
    /// partition. Any other value takes the SBR path, where the
    /// <c>keep-oldest down-if-alone</c> flag is emitted from
    /// <see cref="ClusterOptions.DownIfAlone"/> rather than hard-coded, so the bound
    /// configuration value is actually consumed.
    ///
    /// A <c>downing-provider-class</c> is always installed explicitly: Akka defaults
    /// to <c>NoDowning</c>, under which the downing configuration is inert and
    /// singletons never migrate on a hard crash or partition. Naming the provider is
    /// what activates automatic downing.
    ///
    /// Every duration is rendered via <see cref="DurationHocon"/> in
    /// milliseconds, so sub-second cluster timing values (e.g. a 750ms heartbeat) are
    /// preserved exactly instead of being rounded to whole seconds.
    /// </summary>
    /// <param name="nodeOptions">The node configuration options.</param>
    /// <param name="clusterOptions">The cluster configuration options.</param>
    /// <param name="roles">The list of node roles to configure.</param>
    /// <param name="transportHeartbeat">The transport heartbeat interval.</param>
    /// <param name="transportFailure">The transport failure threshold.</param>
    /// <returns>The Akka HOCON configuration string.</returns>
    public static string BuildHocon(
        NodeOptions nodeOptions,
        ClusterOptions clusterOptions,
        IEnumerable<string> roles,
        TimeSpan transportHeartbeat,
        TimeSpan transportFailure)
    {
        // Simultaneous-cold-start split-brain guard (Gitea #33): when enabled the node must NOT
        // auto-join from its config seeds — Akka's FirstSeedNodeProcess is exactly what races the
        // peer and forms two 1-node clusters. Emit an EMPTY seed list so the ActorSystem comes up
        // unjoined, and ClusterBootstrapCoordinator issues the single reachability-gated
        // JoinSeedNodes (founder self-first / joiner peer-first). Default OFF, so guard-off configs
        // render byte-identical to before.
        var effectiveSeeds = clusterOptions.BootstrapGuard.Enabled
            ? Enumerable.Empty<string>()
            : clusterOptions.SeedNodes;
        var seedNodesStr = string.Join(",", effectiveSeeds.Select(QuoteHocon));
        var rolesStr = string.Join(",", roles.Select(QuoteHocon));

        // auto-down (default): AutoDowning provider — the leader among the reachable
        // members downs the unreachable peer after StableAfter, so a crash of EITHER
        // node fails over to the survivor (dual-active during a real partition is the
        // accepted trade — decision 2026-07-21). Anything else: the SBR provider with
        // the configured active-strategy (keep-oldest), which is partition-safe but
        // cannot survive a crash of the oldest node in a two-node cluster.
        var downingBlock = string.Equals(
            clusterOptions.SplitBrainResolverStrategy, "auto-down", StringComparison.OrdinalIgnoreCase)
            ? $@"downing-provider-class = ""Akka.Cluster.AutoDowning, Akka.Cluster""
        auto-down-unreachable-after = {DurationHocon(clusterOptions.StableAfter)}"
            : $@"downing-provider-class = ""Akka.Cluster.SBR.SplitBrainResolverProvider, Akka.Cluster""
        split-brain-resolver {{
            active-strategy = {QuoteHocon(clusterOptions.SplitBrainResolverStrategy)}
            stable-after = {DurationHocon(clusterOptions.StableAfter)}
            keep-oldest {{
                down-if-alone = {(clusterOptions.DownIfAlone ? "on" : "off")}
            }}
        }}";

        return $@"
audit-telemetry-dispatcher {{
    type = ForkJoinDispatcher
    throughput = 100
    dedicated-thread-pool {{
        thread-count = 2
    }}
}}
akka {{
    extensions = [
        ""Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubExtensionProvider, Akka.Cluster.Tools""
    ]
    actor {{
        provider = cluster
    }}
    remote {{
        dot-netty.tcp {{
            hostname = {QuoteHocon(nodeOptions.NodeHostname)}
            port = {nodeOptions.RemotingPort}
        }}
        transport-failure-detector {{
            heartbeat-interval = {DurationHocon(transportHeartbeat)}
            acceptable-heartbeat-pause = {DurationHocon(transportFailure)}
        }}
    }}
    cluster {{
        seed-nodes = [{seedNodesStr}]
        roles = [{rolesStr}]
        min-nr-of-members = {clusterOptions.MinNrOfMembers}
        {downingBlock}
        failure-detector {{
            heartbeat-interval = {DurationHocon(clusterOptions.HeartbeatInterval)}
            acceptable-heartbeat-pause = {DurationHocon(clusterOptions.FailureDetectionThreshold)}
        }}
        run-coordinated-shutdown-when-down = on
    }}
    coordinated-shutdown {{
        run-by-clr-shutdown-hook = on
        phases.cluster-leave.timeout = 15s
    }}
}}";
    }

    /// <summary>
    /// Renders a <see cref="TimeSpan"/> as a HOCON duration in milliseconds. Akka's
    /// HOCON parser accepts a <c>ms</c> suffix, so emitting whole milliseconds
    /// preserves sub-second configuration exactly — a 750ms heartbeat stays 750ms
    /// rather than being rounded to <c>1s</c> (or, for sub-half-second values,
    /// silently collapsing to a degenerate <c>0s</c>).
    /// </summary>
    private static string DurationHocon(TimeSpan duration)
    {
        return $"{(long)Math.Round(duration.TotalMilliseconds)}ms";
    }

    /// <summary>
    /// Renders a value as a HOCON double-quoted string, escaping backslashes and
    /// double quotes so the resulting token cannot break out of its string literal.
    /// </summary>
    private static string QuoteHocon(string? value)
    {
        var escaped = (value ?? string.Empty)
            .Replace("\\", "\\\\")
            .Replace("\"", "\\\"");
        return $"\"{escaped}\"";
    }

    /// <summary>
    /// Stops the Akka.NET actor system and cleans up resources.
    /// </summary>
    /// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
    /// <returns>A task representing the asynchronous operation.</returns>
    public async Task StopAsync(CancellationToken cancellationToken)
    {
        // Mark this as an intentional shutdown FIRST, so the WhenTerminated
        // watchdog (wired in GetOrCreateActorSystem) does not mistake the
        // CoordinatedShutdown below for an SBR self-down and re-trigger
        // StopApplication on an already-stopping host.
        _stopRequested = true;

        // Dispose auxiliary subscribers (e.g. SiteAuditTelemetryStalledTracker)
        // BEFORE Akka shuts down so their EventStream unsubscribe calls run
        // while the system is still alive. Per-tracker Dispose is wrapped in
        // its own try so a misbehaving subscriber can't sink the shutdown.
        // Snapshot the list inside a lock so a concurrent StartAsync (the
        // test harness sometimes triggers a second start/stop interleaving)
        // can't race the enumeration. Clearing the original list under the
        // same lock leaves the next StartAsync with a clean slate.
        IDisposable[] disposables;
        lock (_trackedDisposables)
        {
            disposables = _trackedDisposables.ToArray();
            _trackedDisposables.Clear();
        }
        foreach (var disposable in disposables)
        {
            try { disposable.Dispose(); }
            catch (Exception ex)
            {
                _logger.LogWarning(ex,
                    "Auxiliary subscriber {Type} threw during shutdown",
                    disposable.GetType().Name);
            }
        }

        if (_actorSystem != null)
        {
            _logger.LogInformation("Shutting down Akka.NET actor system via CoordinatedShutdown...");
            var shutdown = Akka.Actor.CoordinatedShutdown.Get(_actorSystem);
            await shutdown.Run(Akka.Actor.CoordinatedShutdown.ClrExitReason.Instance);
            _logger.LogInformation("Akka.NET actor system shutdown complete.");
        }
    }

    /// <summary>
    /// Builds the list of cluster roles for this node. Site nodes get both "Site"
    /// and a site-specific role (e.g., "site-SiteA") to scope singleton placement.
    /// </summary>
    private List<string> BuildRoles()
    {
        var roles = new List<string> { _nodeOptions.Role };

        if (_nodeOptions.Role.Equals("Site", StringComparison.OrdinalIgnoreCase)
            && !string.IsNullOrEmpty(_nodeOptions.SiteId))
        {
            roles.Add($"site-{_nodeOptions.SiteId}");
        }

        return roles;
    }

    /// <summary>
    /// Registers central-side actors including the CentralCommunicationActor.
    /// Central communication actor routes all 8 message patterns to sites.
    /// </summary>
    private void RegisterCentralActors()
    {
        // Feed this central node's hostname into the local health collector so
        // the CentralHealthReportLoop's report identifies the active node.
        var centralHealthCollector = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.HealthMonitoring.ISiteHealthCollector>();
        centralHealthCollector?.SetNodeHostname(_nodeOptions.NodeHostname);

        // Central→site command transport: gRPC SiteCommandService. This is the ONLY transport since
        // the ClusterClient→gRPC migration's Phase 4 removed the Akka path; the actor no longer
        // chooses a transport, the Host builds it. Constructed once from the DI
        // SitePairChannelProvider and shared across CentralCommunicationActor incarnations (it holds
        // only the shared provider + options); the actor reconciles its per-site channel pairs from
        // the DB refresh loop.
        var siteCommandTransport = new GrpcSiteTransport(
            _serviceProvider.GetRequiredService<SitePairChannelProvider>(),
            _communicationOptions,
            _serviceProvider.GetRequiredService<ILoggerFactory>().CreateLogger<GrpcSiteTransport>());
        _logger.LogInformation("central→site command transport: gRPC (SiteCommandService)");
        var centralCommActor = _actorSystem!.ActorOf(
            Props.Create(() => new CentralCommunicationActor(_serviceProvider, siteCommandTransport)),
            "central-communication");

        // Hand the same actor to the central-hosted gRPC control plane (T1A.2) and open its
        // readiness gate — the gRPC face Asks this exact actor, so both transports resolve to
        // one handler implementation. Mirrors SiteStreamGrpcServer.SetReady on the site side:
        // the service is a DI singleton created before the actor system exists, so the actor
        // arrives here post-construction. Null on a host that did not register the service
        // (e.g. an in-process test harness), so the wiring is a guarded no-op there.
        var centralControlGrpc = _serviceProvider
            .GetService<ZB.MOM.WW.ScadaBridge.Communication.Grpc.CentralControlGrpcService>();
        centralControlGrpc?.SetReady(centralCommActor);
        _logger.LogInformation(
            "CentralControlGrpcService readiness set (service bound: {Bound})",
            centralControlGrpc is not null);

        // Wire up the CommunicationService with the actor reference
        var commService = _serviceProvider.GetService<CommunicationService>();
        commService?.SetCommunicationActor(centralCommActor);

        // Wire up the DebugStreamService with the ActorSystem
        var debugStreamService = _serviceProvider.GetService<DebugStreamService>();
        debugStreamService?.SetActorSystem(_actorSystem!);

        // Wire up the aggregated live alarm cache with the ActorSystem so it can host
        // per-site aggregator actors on the active central node (plan #10, Task 4).
        var siteAlarmLiveCache = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.Communication.SiteAlarmLiveCacheService>();
        siteAlarmLiveCache?.SetActorSystem(_actorSystem!);


        // Management Service — reached IN-PROCESS only, via ManagementActorHolder.
        //
        // This actor used to be registered with the ClusterClientReceptionist as well, for
        // an out-of-cluster CLI that was never built (REQ-HOST-6a). The shipped CLI speaks
        // HTTP Basic to /management, which asks this actor through the holder below
        // (ManagementEndpoints), so the registration had no sender anywhere in the repo —
        // it only advertised a management surface across the cluster-client boundary for
        // free. Removed 2026-07-22 (ClusterClient→gRPC migration, T0.1).
        var mgmtLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<ZB.MOM.WW.ScadaBridge.ManagementService.ManagementActor>();
        var mgmtActor = _actorSystem!.ActorOf(
            Props.Create(() => new ZB.MOM.WW.ScadaBridge.ManagementService.ManagementActor(_serviceProvider, mgmtLogger)),
            "management");
        var mgmtHolder = _serviceProvider.GetRequiredService<ZB.MOM.WW.ScadaBridge.ManagementService.ManagementActorHolder>();
        mgmtHolder.ActorRef = mgmtActor;
        _logger.LogInformation("ManagementActor started at /user/management (in-process access via ManagementActorHolder)");

        // Notification Outbox — cluster singleton so exactly one node owns ingest,
        // the dispatch sweep and the purge loop. Central actors run on the base
        // "Central" role, so the singleton settings are NOT role-scoped (unlike the
        // site singletons, which are scoped to a per-site role).
        var outboxOptions = _serviceProvider
            .GetRequiredService<IOptions<ZB.MOM.WW.ScadaBridge.NotificationOutbox.NotificationOutboxOptions>>().Value;
        var outboxLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<ZB.MOM.WW.ScadaBridge.NotificationOutbox.NotificationOutboxActor>();
        // Central direct-write audit writer for dispatcher attempt
        // + terminal events. Resolved once from the root provider — the writer
        // is a singleton and stateless, opening per-call DI scopes internally
        // to resolve the scoped IAuditLogRepository.
        var outboxAuditWriter = _serviceProvider
            .GetRequiredService<ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.ICentralAuditWriter>();

        // All central singletons below are created through SingletonRegistrar.Start,
        // which pins the actor to the active central node (ClusterSingletonManager), exposes a
        // stable address (ClusterSingletonProxy), and — crucially — registers a PhaseClusterLeave
        // GracefulStop drain task so an in-flight EF write drains before handover (review 01
        // [Medium]: notification-outbox and audit-log-ingest previously had NO drain task). Names
        // are unchanged ({name}-singleton / {name}-proxy). Site singletons use the same
        // registrar with `role: siteRole` (Task 11 / round-2 N5).
        var outbox = SingletonRegistrar.Start(
            _actorSystem!, "notification-outbox",
            Props.Create(() => new ZB.MOM.WW.ScadaBridge.NotificationOutbox.NotificationOutboxActor(
                _serviceProvider,
                outboxOptions,
                outboxAuditWriter,
                outboxLogger)),
            _logger);

        // Hand the outbox proxy to the CentralCommunicationActor so forwarded
        // NotificationSubmit messages from sites are routed to the outbox singleton.
        centralCommActor.Tell(new RegisterNotificationOutbox(outbox.Proxy));

        // Hand the same proxy to the CommunicationService so the Central UI can
        // Ask the outbox actor directly (query, retry, discard, KPIs).
        commService?.SetNotificationOutbox(outbox.Proxy);
        _logger.LogInformation("NotificationOutbox singleton created and registered with CentralCommunicationActor");

        // Audit Log — central singleton mirrors the Notification Outbox
        // pattern. TWO gRPC servers can carry an IngestAuditEvents call and both Ask this
        // proxy directly: the central-hosted CentralControlGrpcService (the production
        // site→central path) and SiteStreamGrpcServer (Communication.Grpc). Each is handed
        // the proxy below; either may be absent on a given node and the wiring no-ops. Even
        // with neither bound the host still brings the singleton up so an in-process test
        // (or a future direct caller) can Ask the proxy without further wiring.
        // IAuditLogRepository is a SCOPED EF Core service, so the singleton
        // actor takes the root IServiceProvider and creates a fresh scope per
        // message (mirroring NotificationOutboxActor). Pre-resolving the
        // repository here would attempt to take a scoped service from the
        // root and fail under DI scope validation.
        var auditIngestLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor>();

        var auditIngest = SingletonRegistrar.Start(
            _actorSystem!, "audit-log-ingest",
            Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor(
                _serviceProvider,
                auditIngestLogger)),
            _logger);

        // Hand the audit-ingest proxy to the central-hosted gRPC control plane so its two
        // ingest RPCs Ask the singleton DIRECTLY. There is deliberately no
        // CentralCommunicationActor relay any more: it re-Asked this same proxy with the same
        // 30 s timeout, so it could only add a hop and latency to every audit batch.
        centralControlGrpc?.SetAuditIngestActor(auditIngest.Proxy);

        // Hand the proxy to the SiteStreamGrpcServer (if registered on this node)
        // so the IngestAuditEvents RPC routes incoming site batches to the singleton.
        // The gRPC server is currently only registered on Site nodes; on a central
        // node this resolves to null and the wiring is a no-op unless a future
        // central-hosted gRPC server (or a real site→central client) is added.
        var grpcServer = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamGrpcServer>();
        grpcServer?.SetAuditIngestActor(auditIngest.Proxy);
        _logger.LogInformation(
            "AuditLogIngestActor singleton created (control-plane bound: {ControlBound}, "
            + "site-stream server bound: {GrpcBound})",
            centralControlGrpc is not null,
            grpcServer is not null);

        // Subscribe the per-site stalled
        // telemetry tracker to the actor system EventStream NOW that the
        // system exists. The tracker mirrors every
        // SiteAuditTelemetryStalledChanged publication (from
        // SiteAuditReconciliationActor — wired in a later bundle) into the
        // AuditCentralHealthSnapshot singleton so the central health surface
        // sees per-site stalled state. The tracker is constructed here rather
        // than in AddAuditLogCentralMaintenance because its ctor needs an
        // ActorSystem, which is not a DI-resolvable singleton — it's owned
        // by this hosted service. The snapshot singleton is resolvable;
        // passing it in seeds the tracker's Apply() so both internal state
        // and the snapshot stay in lock-step.
        var auditCentralSnapshot = _serviceProvider
            .GetService<ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditCentralHealthSnapshot>();
        if (auditCentralSnapshot is not null)
        {
            var stalledTracker = new ZB.MOM.WW.ScadaBridge.AuditLog.Central.SiteAuditTelemetryStalledTracker(
                _actorSystem!, auditCentralSnapshot);
            lock (_trackedDisposables)
            {
                _trackedDisposables.Add(stalledTracker);
            }
            _logger.LogInformation("SiteAuditTelemetryStalledTracker subscribed to EventStream");
        }

        // Site Call Audit — central singleton mirrors the AuditLogIngest
        // and NotificationOutbox patterns. The dual-write transaction routes
        // SiteCalls upserts through AuditLogIngestActor's own scope-per-message
        // ISiteCallAuditRepository resolution, so this singleton is not on the
        // happy-path hot path; it exists so direct-write callers Ask through
        // a stable cluster proxy without further wiring. The central→site
        // Retry/Discard relay now lives in this actor (see the
        // RegisterCentralCommunication wiring below); the reconciliation puller
        // is the remaining deferred direct-write caller.
        // Like AuditLogIngestActor, the actor takes the root IServiceProvider
        // and creates a fresh scope per message because ISiteCallAuditRepository
        // is a scoped EF Core service.
        var siteCallAuditLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<ZB.MOM.WW.ScadaBridge.SiteCallAudit.SiteCallAuditActor>();
        var siteCallAuditOptions = _serviceProvider
            .GetRequiredService<IOptions<ZB.MOM.WW.ScadaBridge.SiteCallAudit.SiteCallAuditOptions>>().Value;

        var siteCallAudit = SingletonRegistrar.Start(
            _actorSystem!, "site-call-audit",
            Props.Create(() => new ZB.MOM.WW.ScadaBridge.SiteCallAudit.SiteCallAuditActor(
                _serviceProvider,
                siteCallAuditOptions,
                siteCallAuditLogger)),
            _logger);

        // Hand the proxy to the CommunicationService so the Central UI can Ask
        // the Site Call Audit actor directly (query, KPIs, detail) — mirrors the
        // SetNotificationOutbox wiring above.
        commService?.SetSiteCallAudit(siteCallAudit.Proxy);

        // Hand the CentralCommunicationActor to the SiteCallAudit
        // actor so it can relay operator Retry/Discard on parked cached calls to
        // the owning site (over the per-site gRPC transport via SiteEnvelope).
        // Mirrors the RegisterAuditIngest / RegisterNotificationOutbox wiring;
        // the message is sent to the singleton proxy so it reaches whichever
        // central node currently hosts the singleton.
        siteCallAudit.Proxy.Tell(
            new ZB.MOM.WW.ScadaBridge.SiteCallAudit.RegisterCentralCommunication(centralCommActor));
        _logger.LogInformation(
            "SiteCallAuditActor singleton created and registered with CentralCommunicationActor");

        // Start the two central-only maintenance
        // singletons that were fully implemented but never instantiated: the
        // daily AuditLog partition-switch purge (AuditLogPurgeActor) and the
        // periodic per-site audit-event reconciliation pull
        // (SiteAuditReconciliationActor). Both mirror the SiteCallAudit /
        // NotificationOutbox singleton pattern above: a ClusterSingletonManager
        // pins the actor to the active central node, a ClusterSingletonProxy
        // gives a stable address, and a PhaseClusterLeave graceful-stop task
        // drains the in-flight tick before handover. Options + the production
        // ISiteEnumerator + IPullAuditEventsClient come from
        // AddAuditLogCentralReconciliationClient (central composition root only).
        // Both actors take the root IServiceProvider and open their own per-tick
        // DI scope because IAuditLogRepository / ISiteRepository are scoped EF
        // Core services.
        var auditPurgeLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogPurgeActor>();
        var auditPurgeOptions = _serviceProvider
            .GetRequiredService<IOptions<ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogPurgeOptions>>();
        var auditLogOptions = _serviceProvider
            .GetRequiredService<IOptions<ZB.MOM.WW.ScadaBridge.AuditLog.Configuration.AuditLogOptions>>();

        SingletonRegistrar.Start(
            _actorSystem!, "audit-log-purge",
            Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogPurgeActor(
                _serviceProvider,
                auditPurgeOptions,
                auditLogOptions,
                auditPurgeLogger)),
            _logger);
        _logger.LogInformation("AuditLogPurgeActor singleton created");

        // SiteAuditReconciliationActor — self-healing fallback puller. Resolves
        // its production ISiteEnumerator (config-DB Site projection) and
        // IPullAuditEventsClient (gRPC) from the central reconciliation-client
        // helper registered in Program.cs.
        var auditReconLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<ZB.MOM.WW.ScadaBridge.AuditLog.Central.SiteAuditReconciliationActor>();
        var auditReconOptions = _serviceProvider
            .GetRequiredService<IOptions<ZB.MOM.WW.ScadaBridge.AuditLog.Central.SiteAuditReconciliationOptions>>();
        var auditReconSites = _serviceProvider
            .GetRequiredService<ZB.MOM.WW.ScadaBridge.AuditLog.Central.ISiteEnumerator>();
        var auditReconClient = _serviceProvider
            .GetRequiredService<ZB.MOM.WW.ScadaBridge.AuditLog.Central.IPullAuditEventsClient>();

        SingletonRegistrar.Start(
            _actorSystem!, "site-audit-reconciliation",
            Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.SiteAuditReconciliationActor(
                auditReconSites,
                auditReconClient,
                _serviceProvider,
                auditReconOptions,
                auditReconLogger)),
            _logger);
        _logger.LogInformation("SiteAuditReconciliationActor singleton created");

        // KPI History — central singleton that periodically samples the
        // Notification Outbox / Site Call Audit point-in-time KPIs into the
        // KpiHistorySamples table and runs the daily retention purge. Mirrors the
        // audit-log-purge singleton pattern above: a ClusterSingletonManager pins
        // the recorder to the active central node, a ClusterSingletonProxy gives a
        // stable address, and a PhaseClusterLeave graceful-stop task drains the
        // in-flight tick before handover. The recorder's sample + purge timers
        // self-schedule in PreStart. Options come from AddKpiHistory (central
        // composition root only). The actor takes the root IServiceProvider and
        // opens its own per-tick DI scope (the KPI repository is a scoped EF Core
        // service), so the 3 ctor args (IServiceProvider, KpiHistoryOptions,
        // ILogger) are resolved here from DI exactly like the other singletons.
        // NOT readiness-gated by design: KPI history is observability/best-effort
        // (it must never gate /health/ready), so kpi-history-recorder is
        // deliberately absent from RequiredSingletonsHealthCheck.
        var kpiHistoryOptions = _serviceProvider
            .GetRequiredService<IOptions<ZB.MOM.WW.ScadaBridge.KpiHistory.KpiHistoryOptions>>().Value;
        var kpiHistoryLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<ZB.MOM.WW.ScadaBridge.KpiHistory.KpiHistoryRecorderActor>();

        SingletonRegistrar.Start(
            _actorSystem!, "kpi-history-recorder",
            Props.Create(() => new ZB.MOM.WW.ScadaBridge.KpiHistory.KpiHistoryRecorderActor(
                _serviceProvider,
                kpiHistoryOptions,
                kpiHistoryLogger)),
            _logger);
        _logger.LogInformation("KpiHistoryRecorderActor singleton created (not readiness-gated)");

        // Notify-and-fetch deploy transport — central singleton that periodically
        // reclaims expired (TTL-elapsed) PendingDeployment staging rows. Mirrors the
        // kpi-history-recorder singleton pattern above: a ClusterSingletonManager pins
        // it to the active central node, a ClusterSingletonProxy gives a stable address,
        // and a PhaseClusterLeave graceful-stop task drains the in-flight tick before
        // handover. The purge timer self-schedules in PreStart. NOT readiness-gated by
        // design: the purge is best-effort hygiene — supersession bounds the table to ≤1
        // pending row per instance and the config-fetch endpoint enforces the TTL on read,
        // so correctness never depends on it; pending-deployment-purge is therefore
        // deliberately absent from RequiredSingletonsHealthCheck. The actor takes the root
        // IServiceProvider and opens its own per-tick DI scope (IDeploymentManagerRepository
        // is a scoped EF Core service).
        var pendingPurgeLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<PendingDeploymentPurgeActor>();
        var pendingPurgeCommunicationOptions =
            _serviceProvider.GetRequiredService<IOptions<CommunicationOptions>>();

        SingletonRegistrar.Start(
            _actorSystem!, "pending-deployment-purge",
            Props.Create(() => new PendingDeploymentPurgeActor(
                _serviceProvider,
                pendingPurgeCommunicationOptions,
                pendingPurgeLogger)),
            _logger);
        _logger.LogInformation("PendingDeploymentPurgeActor singleton created (not readiness-gated)");

        _logger.LogInformation("Central actors registered. CentralCommunicationActor created.");
    }

    /// <summary>
    /// Registers site-specific actors including the Deployment Manager cluster singleton
    /// and the SiteCommunicationActor.
    /// The singleton is scoped to the site-specific cluster role so it runs on exactly
    /// one node within this site's cluster.
    /// </summary>
    private async Task RegisterSiteActorsAsync(CancellationToken cancellationToken)
    {
        var siteRole = $"site-{_nodeOptions.SiteId}";
        var storage = _serviceProvider.GetRequiredService<SiteStorageService>();
        var compilationService = _serviceProvider.GetRequiredService<ScriptCompilationService>();
        var sharedScriptLibrary = _serviceProvider.GetRequiredService<SharedScriptLibrary>();
        var streamManager = _serviceProvider.GetRequiredService<SiteStreamManager>();
        streamManager.Initialize(_actorSystem!);
        var siteRuntimeOptionsValue = _serviceProvider.GetService<IOptions<SiteRuntimeOptions>>()?.Value
            ?? new SiteRuntimeOptions();
        var dmLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<DeploymentManagerActor>();

        // Create DCL Manager Actor for tag subscriptions
        var dclFactory = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.DataConnectionLayer.IDataConnectionFactory>();
        var dclOptions = _serviceProvider.GetService<IOptions<ZB.MOM.WW.ScadaBridge.DataConnectionLayer.DataConnectionOptions>>()?.Value
            ?? new ZB.MOM.WW.ScadaBridge.DataConnectionLayer.DataConnectionOptions();
        IActorRef? dclManager = null;
        if (dclFactory != null)
        {
            var healthCollector = _serviceProvider.GetRequiredService<ZB.MOM.WW.ScadaBridge.HealthMonitoring.ISiteHealthCollector>();
            var siteEventLogger = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.SiteEventLogging.ISiteEventLogger>();
            // The verify-endpoint probe builds an OPC UA ApplicationConfiguration directly,
            // so the manager needs the same deployment-wide OpcUaGlobalOptions the
            // DataConnectionFactory feeds to RealOpcUaClient when creating connections.
            var opcUaGlobalOptions = _serviceProvider
                .GetService<IOptions<ZB.MOM.WW.ScadaBridge.DataConnectionLayer.OpcUaGlobalOptions>>()?.Value
                ?? new ZB.MOM.WW.ScadaBridge.DataConnectionLayer.OpcUaGlobalOptions();
            dclManager = _actorSystem!.ActorOf(
                Props.Create(() => new ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors.DataConnectionManagerActor(
                    dclFactory, dclOptions, healthCollector, siteEventLogger, opcUaGlobalOptions)),
                "dcl-manager");
            _logger.LogInformation("Data Connection Layer manager actor created");
        }

        // Per-node OPC UA certificate-store actor. Created on EVERY
        // site node (NOT a singleton) at a well-known name so the Deployment
        // Manager singleton can fan a trust/remove out to BOTH nodes' PKI stores
        // (node-a + node-b) and keep them in lock-step across failover. It needs
        // the same deployment-wide OpcUaGlobalOptions the DCL manager uses so a
        // trusted cert lands in the exact store RealOpcUaClient validates against.
        var certStoreOpcUaOptions = _serviceProvider
            .GetService<IOptions<ZB.MOM.WW.ScadaBridge.DataConnectionLayer.OpcUaGlobalOptions>>()?.Value
            ?? new ZB.MOM.WW.ScadaBridge.DataConnectionLayer.OpcUaGlobalOptions();
        _actorSystem!.ActorOf(
            Props.Create(() => new ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.CertStoreActor(certStoreOpcUaOptions)),
            ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.CertStoreActor.WellKnownName);
        _logger.LogInformation("Per-node CertStoreActor created at well-known name '{Name}' (T17/D6)",
            ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.CertStoreActor.WellKnownName);

        // Resolve the health collector for the Deployment Manager
        var siteHealthCollector = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.HealthMonitoring.ISiteHealthCollector>();
        siteHealthCollector?.SetNodeHostname(_nodeOptions.NodeHostname);

        // Notify-and-fetch: the deployment config fetcher pulls a deployment's flattened
        // config from central over HTTP. Used by BOTH the active singleton
        // (RefreshDeploymentCommand) AND the standby replication path — the active
        // node now replicates only the deployment id and the standby fetches the config
        // itself, so a large config never crosses the intra-site Akka hop. Resolve once.
        var deploymentConfigFetcher =
            _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment.IDeploymentConfigFetcher>();

        // ONE active-node predicate instance governs the S&F delivery gate and the
        // heartbeat IsActive stamp (SiteCommunicationActor, wired below) — review 02
        // round 2, N1. It also governed SiteReplicationActor's resync authority until
        // LocalDb Phase 2 deleted that actor: the library's snapshot resync merges per row
        // under last-writer-wins and never deletes, so there is no destructive apply left
        // to need an authority check. Null in non-clustered test hosts: the consumers fall
        // back to the shared oldest-Up evaluator, never to a leader check.
        var clusterNodeProvider = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.HealthMonitoring.IClusterNodeProvider>();
        Func<bool>? activeNodeCheck = clusterNodeProvider != null
            ? () => clusterNodeProvider.SelfIsPrimary
            : null;

        // Deployment Manager — role-scoped singleton via SingletonRegistrar
        // (review 01 round-2 N5): previously hand-rolled with bare PoisonPill
        // termination and NO PhaseClusterLeave drain, so in-flight SQLite
        // writes (static overrides, native_alarm_state) could be cut off on
        // graceful failover. Names unchanged: deployment-manager-singleton /
        // deployment-manager-proxy.
        // WP2.6a: the shared cache backing SiteExternalSystemRepository reads
        // (registered singleton by AddSiteRuntime); DeploymentManagerActor invalidates
        // it after an artifact deploy applies external-system changes.
        var externalSystemCache = _serviceProvider
            .GetService<ZB.MOM.WW.ScadaBridge.SiteRuntime.Repositories.ExternalSystemDefinitionCache>();
        var dm = SingletonRegistrar.Start(
            _actorSystem!, "deployment-manager",
            Props.Create(() => new DeploymentManagerActor(
                storage, compilationService, sharedScriptLibrary, streamManager,
                siteRuntimeOptionsValue, dmLogger, dclManager,
                siteHealthCollector, _serviceProvider, null, deploymentConfigFetcher,
                null, null, externalSystemCache)),
            _logger, role: siteRole);
        var dmProxy = dm.Proxy;

        // Site→central transport: a gRPC dial of CentralControlService with a sticky
        // central-a→central-b channel pair, presenting this site's preshared key. This is the ONLY
        // transport since the ClusterClient→gRPC migration's Phase 4 removed the Akka path;
        // StartupValidator guarantees a Site node lists at least one CentralGrpcEndpoint.
        var loggerFactory = _serviceProvider.GetRequiredService<ILoggerFactory>();
        var channelProvider = new CentralChannelProvider(
            _communicationOptions.CentralGrpcEndpoints,
            new StaticSitePskProvider(_communicationOptions.GrpcPsk),
            _nodeOptions.SiteId!,
            _communicationOptions,
            loggerFactory.CreateLogger<CentralChannelProvider>());
        _trackedDisposables.Add(channelProvider);
        ICentralTransport centralTransport = new GrpcCentralTransport(
            channelProvider,
            _communicationOptions,
            loggerFactory.CreateLogger<GrpcCentralTransport>());
        _logger.LogInformation(
            "Site→central transport: gRPC to {Count} central endpoint(s) for site {SiteId}",
            _communicationOptions.CentralGrpcEndpoints.Count, _nodeOptions.SiteId);

        // The ONE routing table for central→site commands, shared by the Akka
        // SiteCommunicationActor (below) and the gRPC SiteCommandGrpcService (SetReady at the end
        // of this method). Its failover seam resolves (dryRun) or performs the graceful Leave via
        // the shared ClusterFailoverCoordinator so the central/site paths cannot drift.
        var siteCommandDispatcher = new SiteCommandDispatcher(
            _nodeOptions.SiteId!,
            dmProxy,
            (role, dryRun) => ZB.MOM.WW.ScadaBridge.Communication.ClusterState.ClusterFailoverCoordinator
                .FailOverOldest(_actorSystem!, role, dryRun)?.ToString());

        // Create SiteCommunicationActor for receiving messages from central. It routes commands
        // through the shared dispatcher; RegisterLocalHandler (below) registers the handlers INTO
        // that dispatcher, so the gRPC command service sees the same registrations. The site→central
        // transport is selected above (default Akka); both are passed in.
        var siteCommActor = _actorSystem.ActorOf(
            Props.Create(() => new SiteCommunicationActor(
                _nodeOptions.SiteId!,
                _communicationOptions,
                dmProxy,
                activeNodeCheck,
                null,
                centralTransport,
                siteCommandDispatcher)),
            "site-communication");

        // Register local handlers with SiteCommunicationActor
        siteCommActor.Tell(new RegisterLocalHandler(LocalHandlerType.Artifacts, dmProxy));

        // Event log handler — cluster singleton so queries always reach the copy that is
        // ACTIVELY BEING WRITTEN: the site_events table IS replicated (LocalDb Phase 1,
        // SiteLocalDbSetup.ReplicatedTables), but replication alone would not save a
        // per-node handler — central queries route through SiteCommandDispatcher to
        // whichever node answers, and a standby's replica sits idle until failover. The
        // singleton is what guarantees queries always read the actively-written copy;
        // replication is what gives the singleton HISTORY to read after it moves —
        // without it, a failover would restart the event log empty at exactly the moment
        // an operator is investigating an incident (WP3.2 design memo, Decision (b)).
        var eventLogQueryService = _serviceProvider.GetService<SiteEventLogging.IEventLogQueryService>();
        if (eventLogQueryService != null)
        {
            var eventLog = SingletonRegistrar.Start(
                _actorSystem, "event-log-handler",
                Props.Create(() => new SiteEventLogging.EventLogHandlerActor(eventLogQueryService)),
                _logger, role: siteRole);
            siteCommActor.Tell(new RegisterLocalHandler(LocalHandlerType.EventLog, eventLog.Proxy));
        }

        // Parked message handler — bridges Akka to StoreAndForwardService
        var storeAndForwardService = _serviceProvider.GetService<StoreAndForwardService>();
        if (storeAndForwardService != null)
        {
            // Initialize SQLite schema and start the retry timer. Must complete before
            // any actor or HTTP handler touches the service. Awaited rather
            // than blocked via GetAwaiter().GetResult() — no thread-pool starvation /
            // sync-context deadlock risk, and exceptions surface as their original type.
            cancellationToken.ThrowIfCancellationRequested();
            await storeAndForwardService.StartAsync();

            // Standby must be passive: only the active site node runs the S&F
            // delivery sweep. Without this gate BOTH nodes deliver the same
            // replicated Pending rows — systematic duplicate external calls and
            // DB writes (arch review 02, Stability). Re-evaluated per sweep
            // tick so failover resumes delivery within one RetryTimerInterval.
            // IClusterNodeProvider.SelfIsPrimary is the canonical "this node is the
            // oldest Up member (singleton host)" check from the cluster-infrastructure
            // fix plan — the shared helper this seam was designed to accept. The provider
            // is resolved once above (the same instance gating the resync authority and
            // heartbeat IsActive stamp — N1). In a non-clustered test host the provider is
            // unregistered, so the gate stays unset and the sweep is ungated (legacy
            // behaviour, preserved).
            if (clusterNodeProvider != null)
            {
                storeAndForwardService.SetDeliveryGate(() => clusterNodeProvider.SelfIsPrimary);
            }

            // Register the store-and-forward delivery handlers so buffered
            // ExternalSystem calls, cached DB writes and notifications are actually
            // delivered by the retry sweep. Without this, every buffered message is
            // persisted but never delivered. Each handler resolves its scoped consumer
            // service in a fresh DI scope — the sweep runs on a timer thread, outside
            // any request scope.
            storeAndForwardService.RegisterDeliveryHandler(
                ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.StoreAndForwardCategory.ExternalSystem,
                async msg =>
                {
                    using var scope = _serviceProvider.CreateScope();
                    return await scope.ServiceProvider
                        .GetRequiredService<ZB.MOM.WW.ScadaBridge.ExternalSystemGateway.ExternalSystemClient>()
                        .DeliverBufferedAsync(msg);
                });
            storeAndForwardService.RegisterDeliveryHandler(
                ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.StoreAndForwardCategory.CachedDbWrite,
                async msg =>
                {
                    using var scope = _serviceProvider.CreateScope();
                    return await scope.ServiceProvider
                        .GetRequiredService<ZB.MOM.WW.ScadaBridge.ExternalSystemGateway.DatabaseGateway>()
                        .DeliverBufferedAsync(msg);
                });
            // Notification Outbox: a buffered notification is no longer delivered by
            // the site over SMTP. "Delivering" it means forwarding it to the central
            // cluster via the SiteCommunicationActor and treating central's
            // NotificationSubmitAck as the outcome (accepted → delivered; not accepted
            // or timeout → throw → transient → keep buffering). Central owns SMTP.
            //
            // Register exactly once. The sentinel guard
            // catches a second registration path that re-introduces the dead
            // site-SMTP handler — see the sentinel's XML doc above for the
            // historical context. Throwing here is intentional: a silent overwrite
            // by a future maintainer would invert the design back to site-side
            // delivery (NotificationForwarder vs. NotificationDeliveryService).
            if (_notificationDeliveryHandlerRegistered)
            {
                throw new InvalidOperationException(
                    "NotificationService-020: A Notification-category store-and-forward "
                    + "delivery handler was already registered. The canonical handler is "
                    + "NotificationForwarder (central-only delivery, post-redesign). "
                    + "If you are re-introducing a second registration path, remove the "
                    + "first one — RegisterDeliveryHandler is last-write-wins per category "
                    + "and a duplicate inverts the design.");
            }
            var notificationForwarder = new ZB.MOM.WW.ScadaBridge.StoreAndForward.NotificationForwarder(
                siteCommActor,
                _nodeOptions.SiteId!,
                _communicationOptions.NotificationForwardTimeout);
            storeAndForwardService.RegisterDeliveryHandler(
                ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.StoreAndForwardCategory.Notification,
                notificationForwarder.DeliverAsync);
            _notificationDeliveryHandlerRegistered = true;
            _logger.LogInformation(
                "Store-and-forward delivery handlers registered (ExternalSystem, CachedDbWrite, Notification)");

            var parkedMessageHandler = _actorSystem.ActorOf(
                Props.Create(() => new ParkedMessageHandlerActor(
                    storeAndForwardService, _nodeOptions.SiteId!)),
                "parked-message-handler");
            siteCommActor.Tell(new RegisterLocalHandler(LocalHandlerType.ParkedMessages, parkedMessageHandler));
        }

        _logger.LogInformation(
            "Site actors registered. DeploymentManager singleton scoped to role={SiteRole}, SiteCommunicationActor created.",
            siteRole);

        // Per-node startup reconciliation. Created on EVERY site node (NOT a
        // singleton) so a standby that was DOWN during a deploy self-heals on its next
        // restart: it reports its local deployed inventory to central via the
        // SiteCommunicationActor Ask, fetches the gap (missing/stale) over HTTP, and
        // guarded-writes it (orphans are logged, never deleted). Requires the HTTP
        // config fetcher; if it is somehow absent the self-heal is skipped (best-effort —
        // replication remains the primary path and the next restart retries).
        // The dmProxy is passed so the pass heals the RUNTIME too (register row 67): each
        // enabled instance it restores gets an EnsureInstanceStartedCommand routed to the
        // singleton, which starts the Instance Actor if the boot-time load missed it. Going
        // through the PROXY (not a local ActorOf) keeps Instance Actors under the singleton
        // regardless of which node's reconcile pass found the gap.
        //
        // Whichever branch is taken is LATCHED onto the site health report as
        // StartupReconcilerActive (deferred-work register row 73). Before that, the skip
        // was a single boot Warning and nothing else: a node that silently lost its entire
        // self-heal path went on reporting itself perfectly healthy forever, and the only
        // evidence was one line scrolled off the top of a log. Central badges the false
        // case at Warning rather than failing the site — replication is still the primary
        // path, so the node is degraded, not down.
        if (deploymentConfigFetcher != null)
        {
            var reconcileLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
                .CreateLogger<SiteReconciliationActor>();
            _actorSystem.ActorOf(
                Props.Create(() => new SiteReconciliationActor(
                    storage,
                    deploymentConfigFetcher,
                    siteCommActor,
                    _nodeOptions.SiteId!,
                    _nodeOptions.NodeName,
                    reconcileLogger,
                    null,
                    null,
                    dmProxy)),
                "site-reconciliation");
            siteHealthCollector?.SetStartupReconcilerActive(true);
            _logger.LogInformation(
                "SiteReconciliationActor created (per-node startup self-heal) for site {SiteId} node {Node}",
                _nodeOptions.SiteId, _nodeOptions.NodeName);
        }
        else
        {
            siteHealthCollector?.SetStartupReconcilerActive(false);

            // Raised from Warning to Error (register row 73). This is a misconfiguration,
            // not a degraded-but-expected state: no configuration makes the fetcher
            // legitimately absent on a site node, and the consequence is that neither the
            // file heal nor the row-67 runtime heal will ever run on this node.
            _logger.LogError(
                "No IDeploymentConfigFetcher available; SiteReconciliationActor not created for "
                + "site {SiteId} node {Node} — startup self-heal is DISABLED (replication "
                + "remains the primary path). Surfaced centrally as StartupReconcilerActive=false.",
                _nodeOptions.SiteId, _nodeOptions.NodeName);

            // Also record it where an operator reviewing the site after the fact will find
            // it — the boot log line is gone by then, but the site event log is queryable
            // from the Central UI. Best-effort: a failed event write must never keep the
            // node from finishing startup, which is why this is fire-and-forget with an
            // observed continuation rather than an await.
            var bootEventLogger = _serviceProvider
                .GetService<ZB.MOM.WW.ScadaBridge.SiteEventLogging.ISiteEventLogger>();
            _ = bootEventLogger?.LogEventAsync(
                    "deployment",
                    "Error",
                    null,
                    "AkkaHostedService",
                    "Startup self-heal disabled: SiteReconciliationActor was not created "
                    + "because no IDeploymentConfigFetcher is registered.",
                    $"Node {_nodeOptions.NodeName} of site {_nodeOptions.SiteId} cannot "
                    + "reconcile its deployed inventory at boot. Config deploys still reach "
                    + "this node via LocalDb replication when it is up; a deploy that lands "
                    + "while it is DOWN will not be recovered on restart.")
                .ContinueWith(
                    t => _logger.LogWarning(
                        t.Exception,
                        "Failed to write the startup-self-heal-disabled site event."),
                    CancellationToken.None,
                    TaskContinuationOptions.OnlyOnFaulted,
                    TaskScheduler.Default);
        }

        // Audit Log — site-side telemetry actor that drains the SQLite
        // Pending queue and pushes to central via IngestAuditEvents. Not a
        // cluster singleton: each site is its own cluster, and the actor reads
        // node-local SQLite (no replication). The Props are bound to the
        // dedicated audit-telemetry-dispatcher (defined in BuildHocon) so a
        // batch SQLite read + gRPC push never contend with the default
        // dispatcher used by hot-path actors.
        //
        // The SiteAuditTelemetryActor takes its
        // collaborators through its constructor, so we resolve them from DI
        // and pass them in via Props.Create rather than relying on a future
        // FactoryProvider. The real site→central client is constructed and
        // wired immediately below: a SiteCommunicationAuditClient (routing through
        // the SiteCommunicationActor's central transport) replaces the DI-default NoOpSiteStreamAuditClient
        // for site roles, without disturbing the rest of this wiring.
        var siteAuditOptions = _serviceProvider
            .GetRequiredService<IOptions<ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry.SiteAuditTelemetryOptions>>();
        var siteAuditQueue = _serviceProvider
            .GetRequiredService<ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.ISiteAuditQueue>();
        // Audit Log follow-up: the production site→central audit
        // push routes through the SiteCommunicationActor's central transport,
        // not the DI-resolved NoOpSiteStreamAuditClient. The NoOp default stays
        // correct for central/test composition roots (no SiteCommunicationActor);
        // a site role wires the real client here so the
        // SQLite Pending backlog actually drains to central. The forward Ask uses
        // AuditForwardTimeout (35 s) — the OUTERMOST rung of the ingest timeout
        // ladder (35 > 30 gRPC/central Ask > 20 actor budget > 15 SQL), so a
        // slow-but-succeeding central write is never acked to a caller that has
        // already given up and re-sent. It used to reuse NotificationForwardTimeout
        // (30 s), which tied it with the rung below.
        ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry.ISiteStreamAuditClient siteAuditClient =
            new ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry.SiteCommunicationAuditClient(
                siteCommActor,
                _communicationOptions.AuditForwardTimeout);
        var siteAuditLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
            .CreateLogger<ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry.SiteAuditTelemetryActor>();

        // Resolve the site-local operation tracking store so the
        // actor can run the combined-telemetry cached-drain in parallel with
        // the audit-only drain. The store is registered by AddSiteRuntime on
        // site composition roots; on central it is intentionally absent and
        // the cached-drain scheduler is never armed (the central side has no
        // outbound cached calls to track). GetService — null when not
        // registered — matches the optional-param contract on the actor ctor.
        var siteTrackingStore = _serviceProvider
            .GetService<ZB.MOM.WW.ScadaBridge.Commons.Interfaces.IOperationTrackingStore>();

        var siteAuditTelemetryProps = Props.Create(() =>
                new ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry.SiteAuditTelemetryActor(
                    siteAuditQueue,
                    siteAuditClient,
                    siteAuditOptions,
                    siteAuditLogger,
                    siteTrackingStore))
            .WithDispatcher("audit-telemetry-dispatcher");
        _actorSystem.ActorOf(siteAuditTelemetryProps, "site-audit-telemetry");
        _logger.LogInformation(
            "SiteAuditTelemetryActor created (dispatcher=audit-telemetry-dispatcher, client={ClientType}, cachedDrain={CachedDrainEnabled})",
            siteAuditClient.GetType().Name,
            siteTrackingStore is not null);

        // Gate gRPC subscriptions until the actor system and SiteStreamManager are
        // initialized.
        //
        // SetReady asserts a deliberately narrow contract. By this point the
        // actor system exists, SiteStreamManager.Initialize has run, and every
        // role actor (SiteCommunicationActor, deployment-manager singleton,
        // the central transport) has been created with ActorOf —
        // creation and the registration Tells are synchronous and strictly ordered.
        // What is NOT guaranteed is completion of each actor's PreStart or the
        // the central transport's initial connection to central: those are
        // intentionally asynchronous. Gating readiness on the central handshake would
        // be wrong — a site must come up and stream locally even while central is
        // briefly unreachable. gRPC readiness therefore guarantees "the site actor
        // graph exists and can accept subscription streams", not "the cluster
        // handshake has completed". Streams opened before SetReady are already
        // rejected by SiteStreamGrpcServer with StatusCode.Unavailable.
        var grpcServer = _serviceProvider.GetService<ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamGrpcServer>();
        // Audit Log: hand the site-local SqliteAuditWriter (which
        // implements ISiteAuditQueue) to the gRPC server so the PullAuditEvents
        // reconciliation RPC can serve central's pulls. Both the writer and the
        // gRPC server are singletons — wiring this here keeps the dependency
        // direction one-way (Host knows both; Communication doesn't reach back
        // into AuditLog).
        grpcServer?.SetSiteAuditQueue(siteAuditQueue);
        // Site Call Audit: hand the site-local OperationTrackingStore to
        // the gRPC server so the PullSiteCalls reconciliation RPC can serve
        // central's self-heal pulls. siteTrackingStore is resolved above with
        // GetService — present on site composition roots, null on central — so
        // wire the seam only when the store exists. Like SetSiteAuditQueue, both
        // the store and the gRPC server are singletons; wiring here keeps the
        // dependency direction one-way (Host knows both; Communication doesn't
        // reach back into SiteRuntime).
        if (siteTrackingStore is not null)
        {
            grpcServer?.SetOperationTrackingStore(siteTrackingStore);
        }
        grpcServer?.SetReady(_actorSystem!);

        // Site command plane: hand the shared dispatcher to the gRPC command service and flip it
        // ready at the same point as the streaming server — the actor graph exists and the local
        // handlers have been registered into the dispatcher, so it can route commands now.
        var commandService = _serviceProvider
            .GetService<ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteCommandGrpcService>();
        commandService?.SetReady(siteCommandDispatcher);
    }
}