lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs

using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
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.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Configuration.Validation;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;

namespace ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;

/// <summary>
/// Cluster-singleton admin operations actor. Owns the "snapshot the live-edit state and start
/// a deployment" workflow plus (eventually) all mutating live-edit ops invoked by the admin UI.
/// Routed to via <see cref="Commons.Interfaces.IAdminOperationsClient"/> from anywhere in the cluster.
/// </summary>
public sealed class AdminOperationsActor : ReceiveActor
{
    private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
    private readonly IActorRef _coordinator;
    private readonly IReadOnlyDictionary<string, IDriverProbe> _probesByType;
    private readonly TagConfigValidationMode _tagConfigValidationMode;

    /// <summary>
    ///     The coordinator's apply deadline. A <c>Dispatching</c> row with no expected-ack rows that is older
    ///     than this is a lost dispatch, closed by the in-flight check (fix round 1 / I7).
    /// </summary>
    private readonly TimeSpan _applyDeadline;

    /// <summary>
    ///     Where central→node commands are handed off (per-cluster mesh), or
    ///     <see langword="null"/> to publish on the mediator directly.
    /// </summary>
    private readonly IActorRef? _meshRouter;

    private readonly ILoggingAdapter _log = Context.GetLogger();

    /// <summary>The StartDeployment whose compose is running off-actor, or null. See <see cref="HandleStartDeployment"/>.</summary>
    private StartDeployment? _composeInFlight;

    /// <summary>Creates actor props for the admin operations actor.</summary>
    /// <param name="dbFactory">Factory for creating config database contexts.</param>
    /// <param name="coordinator">Reference to the deployment coordinator actor.</param>
    /// <param name="probes">Driver probes registered in DI; keyed by DriverType (case-insensitive).</param>
    /// <param name="tagConfigValidationMode">Deploy-gate TagConfig strictness mode (Warn default | Error opt-in).</param>
    /// <param name="meshRouter">
    ///     Central comm actor commands are handed to, or <see langword="null"/> to publish on the
    ///     DistributedPubSub mediator directly (the behaviour before the per-cluster mesh).
    /// </param>
    /// <param name="applyDeadline">The coordinator's apply deadline; defaults to <see cref="DeployTimings.ApplyDeadline"/> (the coordinator's default source).</param>
    /// <returns>Props configured to create an AdminOperationsActor.</returns>
    public static Props Props(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        IActorRef coordinator,
        IEnumerable<IDriverProbe> probes,
        TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn,
        IActorRef? meshRouter = null,
        TimeSpan? applyDeadline = null) =>
        Akka.Actor.Props.Create(() =>
            new AdminOperationsActor(dbFactory, coordinator, probes, tagConfigValidationMode, meshRouter, applyDeadline));

    /// <summary>Initializes a new instance of the AdminOperationsActor.</summary>
    /// <param name="dbFactory">Factory for creating config database contexts.</param>
    /// <param name="coordinator">Reference to the deployment coordinator actor.</param>
    /// <param name="probes">Driver probes registered in DI; keyed by DriverType (case-insensitive).</param>
    /// <param name="tagConfigValidationMode">Deploy-gate TagConfig strictness mode (Warn default | Error opt-in).</param>
    /// <param name="meshRouter">Central comm actor, or <see langword="null"/> for the mediator.</param>
    /// <param name="applyDeadline">The coordinator's apply deadline; defaults to <see cref="DeployTimings.ApplyDeadline"/> (the coordinator's default source).</param>
    public AdminOperationsActor(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        IActorRef coordinator,
        IEnumerable<IDriverProbe> probes,
        TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn,
        IActorRef? meshRouter = null,
        TimeSpan? applyDeadline = null)
    {
        // The same source the coordinator's default reads (ConfigPublishCoordinator.DefaultApplyDeadline IS
        // DeployTimings.ApplyDeadline), so the lost-dispatch closure and the coordinator's own deadline agree.
        _applyDeadline = applyDeadline ?? DeployTimings.ApplyDeadline;
        _dbFactory = dbFactory;
        _coordinator = coordinator;
        _probesByType = probes.ToDictionary(p => p.DriverType, StringComparer.OrdinalIgnoreCase);
        _tagConfigValidationMode = tagConfigValidationMode;
        _meshRouter = meshRouter;

        // Synchronous receives: the compose and the probe run off-actor and come back as messages, so
        // the mailbox is never suspended for the length of a DB snapshot or a PLC connect timeout.
        Receive<StartDeployment>(HandleStartDeployment);
        Receive<DeployComposed>(HandleDeployComposed);
        Receive<DeployComposeFailed>(HandleDeployComposeFailed);
        Receive<TestDriverConnect>(HandleTestDriverConnect);
        ReceiveAsync<RestartDriver>(HandleRestartDriverAsync);
        ReceiveAsync<ReconnectDriver>(HandleReconnectDriverAsync);
        Receive<AcknowledgeAlarmCommand>(HandleAcknowledgeAlarm);
        Receive<ShelveAlarmCommand>(HandleShelveAlarm);
    }

    /// <summary>
    ///     The reply text for an unexpected server-side failure (R03:4.1): a fixed sentence plus the request's
    ///     correlation id, never <c>ex.Message</c> — a SqlException or UriFormatException message carries
    ///     connection-string fragments and host names, and the reply reaches any ConfigEditor browser session.
    ///     The detail is in the central log under the same correlation id.
    /// </summary>
    /// <param name="operation">The admin operation that failed.</param>
    /// <param name="correlationId">The request's correlation token, as logged alongside the exception.</param>
    /// <returns>The operator-facing message.</returns>
    internal static string ServerFault(string operation, object correlationId) =>
        $"{operation} failed on the server; see the central node log for correlation {correlationId}.";

    /// <summary>
    ///     Hands a central→node command to the mesh transport router, or publishes it directly when
    ///     no router is wired.
    /// </summary>
    /// <remarks>
    ///     The fallback keeps ~12 existing tests (and any node wired before per-cluster mesh)
    ///     on exactly the pre-mesh behaviour. The DistributedPubSub branch behind it is
    ///     <b>retained</b>, not scheduled for deletion — an earlier note here promised Phase 6 would
    ///     remove it, which was deliberately reversed (<c>docs/Configuration.md</c> § "The compiled
    ///     transport-mode defaults were deliberately NOT flipped"): DPS is still the correct and
    ///     default transport for a single-mesh topology; gRPC is what crosses a mesh boundary.
    /// </remarks>
    /// <param name="topic">The legacy DistributedPubSub topic.</param>
    /// <param name="message">The command.</param>
    private void Dispatch(string topic, object message)
    {
        if (_meshRouter is not null) _meshRouter.Tell(new MeshCommand(topic, message));
        else DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(topic, message));
    }

    /// <summary>
    ///     AdminUI Acknowledge path. Maps the control-plane command to a
    ///     <see cref="AlarmCommand"/> (<c>Operation = "Acknowledge"</c>) and publishes it onto the
    ///     cluster <c>alarm-commands</c> topic. The broadcast lands on every driver node's
    ///     <c>ScriptedAlarmHostActor</c>; only the node owning the alarm acts (ownership filter), so the
    ///     admin singleton needs no knowledge of placement. Synchronous — the publish is fire-and-forget
    ///     via the mediator, so there is no awaitable work and no DB write.
    /// </summary>
    private void HandleAcknowledgeAlarm(AcknowledgeAlarmCommand msg)
    {
        var replyTo = Sender;
        try
        {
            var alarmCmd = new AlarmCommand(
                AlarmId: msg.AlarmId,
                Operation: "Acknowledge",
                User: msg.User,
                Comment: msg.Comment,
                UnshelveAtUtc: null);

            Dispatch(AlarmCommandsTopic.Name, alarmCmd);

            _log.Info("AdminOps: Acknowledge published for alarm {AlarmId} by {User}", msg.AlarmId, msg.User);
            replyTo.Tell(new AcknowledgeAlarmResult(true, null, msg.CorrelationId));
        }
        catch (Exception ex)
        {
            _log.Error(ex, "AdminOps: Acknowledge failed for alarm {AlarmId} (correlation {CorrelationId})", msg.AlarmId, msg.CorrelationId);
            replyTo.Tell(new AcknowledgeAlarmResult(false, ServerFault("Acknowledge", msg.CorrelationId), msg.CorrelationId));
        }
    }

    /// <summary>
    ///     AdminUI Shelve / Unshelve path. Maps <see cref="ShelveAlarmCommand.Kind"/> to the matching
    ///     <see cref="AlarmCommand"/> operation (<c>OneShotShelve</c> / <c>TimedShelve</c> /
    ///     <c>Unshelve</c>), threading <see cref="ShelveAlarmCommand.UnshelveAtUtc"/> for the timed kind,
    ///     and publishes onto the cluster <c>alarm-commands</c> topic. Ownership filtering happens on the
    ///     owning node exactly as for Acknowledge.
    /// </summary>
    private void HandleShelveAlarm(ShelveAlarmCommand msg)
    {
        var replyTo = Sender;
        try
        {
            var (operation, unshelveAt) = msg.Kind switch
            {
                ShelveKind.OneShot => ("OneShotShelve", (DateTime?)null),
                ShelveKind.Timed => ("TimedShelve", msg.UnshelveAtUtc),
                ShelveKind.Unshelve => ("Unshelve", (DateTime?)null),
                _ => throw new ArgumentOutOfRangeException(nameof(msg), msg.Kind, "Unknown shelve kind."),
            };

            // TimedShelve requires an unshelve instant — the engine rejects it otherwise. Guard here so
            // the AdminUI gets an immediate, attributable failure instead of a silently-dropped command.
            if (msg.Kind == ShelveKind.Timed && unshelveAt is null)
            {
                replyTo.Tell(new ShelveAlarmResult(false, "TimedShelve requires UnshelveAtUtc.", msg.CorrelationId));
                return;
            }

            var alarmCmd = new AlarmCommand(
                AlarmId: msg.AlarmId,
                Operation: operation,
                User: msg.User,
                Comment: msg.Comment,
                UnshelveAtUtc: unshelveAt);

            Dispatch(AlarmCommandsTopic.Name, alarmCmd);

            _log.Info("AdminOps: {Operation} published for alarm {AlarmId} by {User}", operation, msg.AlarmId, msg.User);
            replyTo.Tell(new ShelveAlarmResult(true, null, msg.CorrelationId));
        }
        catch (Exception ex)
        {
            _log.Error(ex, "AdminOps: Shelve ({Kind}) failed for alarm {AlarmId} (correlation {CorrelationId})", msg.Kind, msg.AlarmId, msg.CorrelationId);
            replyTo.Tell(new ShelveAlarmResult(false, ServerFault("Shelve", msg.CorrelationId), msg.CorrelationId));
        }
    }

    /// <summary>
    ///     <c>StartDeployment</c>. The only serialised part is the actor-local <see cref="_composeInFlight"/>
    ///     flag; the DB in-flight check, the snapshot read, validation, gzip and the row write all run on the
    ///     thread pool and come back as a <see cref="DeployComposed"/> / <see cref="DeployComposeFailed"/>
    ///     message, so a slow or blocked ConfigDb never suspends the mailbox (R03:3.1; fix round 1 moved the
    ///     in-flight query off the actor thread too) and alarm acks keep flowing while it runs.
    /// </summary>
    /// <remarks>
    ///     <see cref="_composeInFlight"/> is what keeps the check correct once it is off-actor: it is set
    ///     BEFORE the off-actor work starts and cleared when the outcome message lands (both branches), so a
    ///     second request is refused in memory for the whole check-plus-compose and two artifacts can never
    ///     race for one coordinator.
    /// </remarks>
    private void HandleStartDeployment(StartDeployment msg)
    {
        var replyTo = Sender;
        if (_composeInFlight is { } inFlightRequest)
        {
            replyTo.Tell(new StartDeploymentResult(
                StartDeploymentOutcome.AnotherDeploymentInFlight,
                DeploymentId: null,
                RevisionHash: null,
                Message: $"A deployment requested by {inFlightRequest.CreatedBy} is still being composed.",
                msg.CorrelationId));
            return;
        }

        _composeInFlight = msg;
        // Captured here: Context/Sender are not available on the thread-pool thread.
        var sourceNode = Akka.Cluster.Cluster.Get(Context.System).SelfAddress.Host ?? "unknown";
        var self = Self;
        Task.Run(async () =>
            {
                var refusal = await CheckInFlightAsync(msg).ConfigureAwait(false);
                return refusal ?? await ComposeDeploymentAsync(msg, sourceNode).ConfigureAwait(false);
            })
            .PipeTo(self,
                success: composed => new DeployComposed(msg, replyTo, composed.Result, composed.Dispatch),
                failure: ex => new DeployComposeFailed(msg, replyTo, ex));
    }

    /// <summary>
    ///     The DB half of the in-flight check, off the actor thread. Refuses while a deployment is in flight —
    ///     keeping the coordinator's state unambiguous — EXCEPT a lost dispatch, which it closes first.
    /// </summary>
    /// <param name="msg">The request.</param>
    /// <returns>The refusal, or null to proceed.</returns>
    /// <remarks>
    ///     <para>
    ///         <b>Lost dispatch (fix round 1 / I7).</b> A <c>Dispatching</c> row with ZERO
    ///         <c>NodeDeploymentState</c> rows was never taken by the coordinator (its <c>HandleDispatch</c>
    ///         seeds the expected set and flips the status in one <c>SaveChanges</c>). The coordinator's PreStart
    ///         recovers such a row — but only one that already existed when it started. A row committed AFTER
    ///         PreStart whose <c>Tell</c> never arrived (this actor died between the compose's commit and
    ///         <see cref="HandleDeployComposed"/>) has no owner at all, and refused every later deploy forever.
    ///         Once it is past the apply deadline nothing can still be legitimately dispatching it, so it is
    ///         closed <c>TimedOut</c> with a reason and the new deploy proceeds. A late <c>DispatchDeployment</c>
    ///         for it is ignored by the coordinator's terminal-state guard.
    ///     </para>
    ///     <para>
    ///         One indexed projection (<c>IX_Deployment_Status</c>) plus, only for a Dispatching row past the
    ///         deadline, a count and a tracked update.
    ///     </para>
    /// </remarks>
    private async Task<ComposeOutcome?> CheckInFlightAsync(StartDeployment msg)
    {
        await using var db = await _dbFactory.CreateDbContextAsync().ConfigureAwait(false);
        var inflight = await db.Deployments
            .AsNoTracking()
            .Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks)
            .OrderByDescending(d => d.CreatedAtUtc)
            .Select(d => new { d.DeploymentId, d.Status, d.CreatedAtUtc })
            .ToListAsync()
            .ConfigureAwait(false);

        foreach (var row in inflight)
        {
            if (row.Status == DeploymentStatus.Dispatching
                && DateTime.UtcNow - row.CreatedAtUtc > _applyDeadline
                && !await db.NodeDeploymentStates.AnyAsync(s => s.DeploymentId == row.DeploymentId).ConfigureAwait(false)
                && await CloseLostDispatchAsync(db, row.DeploymentId).ConfigureAwait(false))
            {
                continue;
            }

            return new ComposeOutcome(
                new StartDeploymentResult(
                    StartDeploymentOutcome.AnotherDeploymentInFlight,
                    DeploymentId: new DeploymentId(row.DeploymentId),
                    RevisionHash: null,
                    Message: $"Deployment {row.DeploymentId:N} is still in flight.",
                    msg.CorrelationId),
                Dispatch: null);
        }
        return null;
    }

    /// <summary>Closes a lost dispatch as TimedOut, unless it moved on meanwhile (optimistic, on its RowVersion).</summary>
    /// <param name="db">The context.</param>
    /// <param name="deploymentId">The row.</param>
    /// <returns><see langword="true"/> when the row is now closed.</returns>
    private async Task<bool> CloseLostDispatchAsync(OtOpcUaConfigDbContext db, Guid deploymentId)
    {
        var stamp = await db.Deployments
            .AsNoTracking()
            .Where(d => d.DeploymentId == deploymentId && d.Status == DeploymentStatus.Dispatching)
            .Select(d => new { d.RowVersion })
            .FirstOrDefaultAsync()
            .ConfigureAwait(false);
        if (stamp is null) return true;   // no longer Dispatching — not in flight as far as this check goes

        // Stub update: the row's ArtifactBlob is never read just to change its status.
        var stub = new Deployment
        {
            DeploymentId = deploymentId, RevisionHash = string.Empty, CreatedBy = string.Empty,
            RowVersion = stamp.RowVersion,
            Status = DeploymentStatus.TimedOut,
            FailureReason =
                "Dispatch was lost before any node was expected to ack (the coordinator was never told, or never " +
                "took it) and the apply deadline has passed; closed by the next StartDeployment. Redeploy.",
        };
        var entry = db.Deployments.Attach(stub);
        entry.Property(d => d.Status).IsModified = true;
        entry.Property(d => d.FailureReason).IsModified = true;
        try
        {
            await db.SaveChangesAsync().ConfigureAwait(false);
        }
        catch (DbUpdateConcurrencyException)
        {
            // The coordinator took it after all (status flipped, RowVersion moved): it is genuinely in flight.
            return false;
        }
        _log.Warning(
            "StartDeployment: closed lost dispatch {DeploymentId} as TimedOut — Dispatching with no expected-ack rows " +
            "past the {Deadline} apply deadline (the coordinator never took it)", deploymentId, _applyDeadline);
        return true;
    }

    private void HandleDeployComposed(DeployComposed msg)
    {
        _composeInFlight = null;
        // The row is committed (the compose task completed), so the coordinator can be told now; it reads
        // the row back by id when it seeds the expected-ack set.
        if (msg.Dispatch is { } dispatch) _coordinator.Tell(dispatch);
        msg.ReplyTo.Tell(msg.Result);
    }

    private void HandleDeployComposeFailed(DeployComposeFailed msg)
    {
        _composeInFlight = null;
        _log.Error(msg.Error, "StartDeployment failed for {CreatedBy} (correlation {CorrelationId})", msg.Request.CreatedBy, msg.Request.CorrelationId);
        msg.ReplyTo.Tell(new StartDeploymentResult(
            StartDeploymentOutcome.Rejected,
            DeploymentId: null,
            RevisionHash: null,
            Message: ServerFault("StartDeployment", msg.Request.CorrelationId),
            msg.Request.CorrelationId));
    }

    /// <summary>Outcome of the off-actor compose: the reply to send, and the dispatch to issue when Accepted.</summary>
    private sealed record ComposeOutcome(StartDeploymentResult Result, DispatchDeployment? Dispatch);

    private sealed record DeployComposed(
        StartDeployment Request, IActorRef ReplyTo, StartDeploymentResult Result, DispatchDeployment? Dispatch);

    private sealed record DeployComposeFailed(StartDeployment Request, IActorRef ReplyTo, Exception Error);

    /// <summary>
    ///     The deploy compose, off the actor thread. Touches no actor state: everything it needs is in
    ///     <paramref name="msg"/>, <paramref name="sourceNode"/> and the readonly fields captured at
    ///     construction (<see cref="_dbFactory"/>, <see cref="_tagConfigValidationMode"/>, <see cref="_log"/> —
    ///     Akka's logging adapter is thread-safe).
    /// </summary>
    private async Task<ComposeOutcome> ComposeDeploymentAsync(StartDeployment msg, string sourceNode)
    {
        await using var db = await _dbFactory.CreateDbContextAsync();

        // Full pre-seal gate: run the complete DraftValidator (one pass, all rules) and reject
        // on ANY validation error. The config is now canonical — the company overlay loader emits
        // canonical EquipmentIds and the seed is clean — so every rule (UNS segments, EquipmentId
        // derivation, cross-cluster/namespace binding, driver-namespace compat, signal collisions,
        // …) gates the deploy. A green build in this repo does not prove the config is valid; this
        // is the last guard before a bad address space (or a non-derived EquipmentId) ships.
        // NOTE: a BadDuplicateExternalIdentifier rejection from the reservation pre-flight can
        // (rarely) be a false positive if an external-id reservation release has not yet been
        // committed/visible when the snapshot is read — operators seeing a spurious one should
        // check ExternalIdReservation state before re-submitting.
        var draft = await DraftSnapshotFactory.FromConfigDbAsync(db);
        var errors = DraftValidator.Validate(draft).ToList();

        // Cluster-topology guard. The SQL
        // CK_ServerCluster_RedundancyMode_NodeCount CHECK enforces the (NodeCount, RedundancyMode)
        // pair on the row itself, but it cannot see the per-node ClusterNode.Enabled flag — an
        // operator can disable a node (effective enabled-count = 1) while leaving RedundancyMode at
        // Hot/Warm and the constraint stays green, which would boot the runtime into an
        // InvalidTopology band. ValidateClusterTopology catches that drift; it is not part of
        // DraftValidator.Validate, so it runs separately here over the snapshot's cluster/node rows.
        // Errors fold into the same reject summary alongside the snapshot rules so a deploy failing
        // either check is rejected with both sets of messages. ClusterId-ordered for a deterministic summary.
        // R03:3.2 — both come off the SAME snapshot read as every other table (one SNAPSHOT
        // transaction), not a second query that could observe a later edit.
        var clusters = draft.ServerClusters.OrderBy(c => c.ClusterId, StringComparer.Ordinal).ToList();
        var nodesByCluster = draft.ClusterNodes
            .GroupBy(n => n.ClusterId, StringComparer.Ordinal)
            .ToDictionary(g => g.Key, g => g.ToList(), StringComparer.Ordinal);
        foreach (var cluster in clusters)
        {
            var nodes = nodesByCluster.TryGetValue(cluster.ClusterId, out var ns)
                ? (IReadOnlyList<ClusterNode>)ns
                : [];
            errors.AddRange(DraftValidator.ValidateClusterTopology(cluster, nodes));
        }

        // Deploy-time TagConfig strictness inspection over equipment tags. The
        // driver-typed EquipmentTagConfigInspector surfaces every blob the driver's own reader would
        // REJECT — a present-but-invalid enum value, an absent or malformed address field, a
        // structurally-unparseable TagConfig — each of which makes the driver skip the tag so it reads
        // BadNodeIdUnknown forever. (The runtime used to silently default a typo'd enum;
        // it now rejects the tag, so these warnings describe rejections.) In Warn mode
        // (default) these are non-blocking (logged + appended to the accepted result Message); in Error
        // mode they fold into the reject list so a typo'd config cannot be re-deployed until fixed.
        // Running servers are untouched either way — the gate only sees re-deploys.
        // v3: a raw Tag's driver is resolved via its Device (Tag no longer carries EquipmentId /
        // DriverInstanceId). Inspect EVERY raw tag's TagConfig against its device's driver type.
        var driverTypeById = draft.DriverInstances
            .GroupBy(d => d.DriverInstanceId, StringComparer.Ordinal)
            .ToDictionary(g => g.Key, g => g.First().DriverType, StringComparer.Ordinal);
        var driverByDevice = draft.Devices
            .GroupBy(d => d.DeviceId, StringComparer.Ordinal)
            .ToDictionary(g => g.Key, g => g.First().DriverInstanceId, StringComparer.Ordinal);
        var deviceConfigById = draft.Devices
            .GroupBy(d => d.DeviceId, StringComparer.Ordinal)
            .ToDictionary(g => g.Key, g => g.First().DeviceConfig, StringComparer.Ordinal);
        var tagConfigWarnings = new List<(string TagId, string Text)>();
        foreach (var t in draft.Tags)
        {
            if (!driverByDevice.TryGetValue(t.DeviceId, out var driverInstanceId)) continue;
            if (!driverTypeById.TryGetValue(driverInstanceId, out var driverType)) continue;
            foreach (var w in EquipmentTagConfigInspector.Inspect(
                driverType, t.TagConfig, deviceConfigById.GetValueOrDefault(t.DeviceId)))
                tagConfigWarnings.Add((t.TagId, $"tag '{t.TagId}': {w}"));
        }
        if (tagConfigWarnings.Count > 0)
        {
            if (_tagConfigValidationMode == TagConfigValidationMode.Error)
                errors.AddRange(tagConfigWarnings.Select(x => new ValidationError("TagConfigInvalid", x.Text, x.TagId)));
            else
                foreach (var x in tagConfigWarnings)
                    _log.Warning("StartDeployment: TagConfig warning (Warn mode, non-blocking): {Warning}", x.Text);
        }

        if (errors.Count > 0)
        {
            var summary = string.Join("; ", errors.Select(e => $"[{e.Code}] {e.Message}"));
            _log.Warning("StartDeployment rejected ({Count} validation error(s)): {Summary}",
                errors.Count, summary);
            return new ComposeOutcome(
                new StartDeploymentResult(
                    StartDeploymentOutcome.Rejected,
                    DeploymentId: null,
                    RevisionHash: null,
                    Message: summary,
                    msg.CorrelationId),
                Dispatch: null);
        }

        // Warn-only compile-cost guardrail (NEVER blocks). Each genuinely-compiled (non-passthrough)
        // script costs ~1.66 MiB RSS per node to materialize via Roslyn (measured post-A0; design doc
        // 2026-06-07). Passthrough "mirror" scripts compile to ~nothing, so they are excluded; distinct
        // sources are counted (the compile cache keys on source, so duplicates collapse to one unit).
        // This only surfaces the estimate to the operator — the DraftValidator gate above is the hard
        // reject; this advisory rides in the Accepted Message so the UI can show it.
        // advisory is declared outside the inner try so the Accepted reply below can still read it even
        // when the guardrail throws. A failed estimate must NEVER block a valid deploy — the caller's
        // failure branch is reserved for genuine seal/save failures.
        string? advisory = null;
        try
        {
            const double PerScriptMiB = 1.66; // measured post-A0 per-script RSS (design doc 2026-06-07)
            // Projected from the snapshot already in hand — Script.SourceCode used to be read three times.
            var compiled = draft.Scripts
                .Select(s => s.SourceCode)
                .Where(src => !PassthroughScript.TryMatch(src, out _))
                .Distinct(StringComparer.Ordinal)
                .Count();
            if (compiled > 0)
            {
                var estMiB = compiled * PerScriptMiB;
                _log.Warning(
                    "StartDeployment: {Compiled} script(s) will compile (~{EstMiB:F0} MiB RSS per node); ensure node mem_limit covers it",
                    compiled, estMiB);
                advisory = $"{compiled} script(s) will compile (~{estMiB:F0} MiB/node)";
            }
        }
        catch (Exception ex)
        {
            // Guardrail is advisory-only — a failed estimate must never block a valid deploy.
            _log.Warning(ex, "StartDeployment: script compile-cost estimate failed; advisory skipped");
            advisory = null;
        }

        // Composed from the validated snapshot, never from a second read: an edit that lands between
        // validation and here is simply not in this artifact (R03:3.2).
        var artifact = ConfigComposer.Compose(draft);
        var deploymentId = DeploymentId.NewId();
        var revHash = RevisionHash.Parse(artifact.RevisionHash);

        db.Deployments.Add(new Deployment
        {
            DeploymentId = deploymentId.Value,
            RevisionHash = artifact.RevisionHash,
            Status = DeploymentStatus.Dispatching,
            CreatedBy = msg.CreatedBy,
            ArtifactBlob = artifact.Blob,
        });

        // Marker ConfigEdit row so the audit timeline shows the deployment snapshot.
        db.ConfigEdits.Add(new ConfigEdit
        {
            EntityType = "Deployment",
            EntityId = deploymentId.Value,
            FieldsJson = $"{{\"revisionHash\":\"{artifact.RevisionHash}\",\"sizeBytes\":{artifact.Blob.Length}}}",
            EditedBy = msg.CreatedBy,
            SourceNode = sourceNode,
        });

        await db.SaveChangesAsync();

        // Accepted message = the compile-cost advisory (if any) + any Warn-mode TagConfig warnings, so
        // the operator sees the non-blocking strictness findings on a successful deploy.
        var messageParts = new List<string>();
        if (advisory is not null) messageParts.Add(advisory);
        messageParts.AddRange(tagConfigWarnings.Select(x => x.Text));
        var acceptedMessage = messageParts.Count > 0 ? string.Join("; ", messageParts) : null;

        return new ComposeOutcome(
            new StartDeploymentResult(
                StartDeploymentOutcome.Accepted,
                deploymentId,
                revHash,
                Message: acceptedMessage,
                msg.CorrelationId),
            new DispatchDeployment(deploymentId, revHash, msg.CorrelationId));
    }

    /// <summary>
    ///     Test Connect runs on the thread pool and pipes its reply straight to the requester: a probe
    ///     against a black-holed PLC (60 s clamp) must not suspend the singleton's mailbox (R03:3.1). Every
    ///     path inside the task yields a <see cref="TestDriverConnectResult"/>, so the requester never
    ///     sees a <c>Status.Failure</c>.
    /// </summary>
    private void HandleTestDriverConnect(TestDriverConnect msg)
    {
        var replyTo = Sender;
        if (!_probesByType.TryGetValue(msg.DriverType, out var probe))
        {
            replyTo.Tell(new TestDriverConnectResult(
                false,
                $"No probe registered for driver type '{msg.DriverType}'.",
                null,
                msg.CorrelationId));
            return;
        }

        Task.Run(() => ProbeAsync(probe, msg)).PipeTo(replyTo);
    }

    private async Task<TestDriverConnectResult> ProbeAsync(IDriverProbe probe, TestDriverConnect msg)
    {
        var clampedSec = Math.Clamp(msg.TimeoutSeconds, 1, 60);
        var timeout = TimeSpan.FromSeconds(clampedSec);
        using var cts = new CancellationTokenSource(timeout);

        var sw = System.Diagnostics.Stopwatch.StartNew();
        try
        {
            var result = await probe.ProbeAsync(msg.ConfigJson, timeout, cts.Token);
            sw.Stop();
            return new TestDriverConnectResult(
                result.Ok,
                result.Message,
                result.Ok ? sw.Elapsed.TotalMilliseconds : (double?)null,
                msg.CorrelationId);
        }
        catch (OperationCanceledException)
        {
            return new TestDriverConnectResult(
                false,
                $"Probe timed out after {clampedSec}s.",
                null,
                msg.CorrelationId);
        }
        catch (Exception ex)
        {
            _log.Error(ex, "Probe for {DriverType} threw (correlation {CorrelationId})", msg.DriverType, msg.CorrelationId);
            return new TestDriverConnectResult(
                false,
                ServerFault("Test Connect", msg.CorrelationId),
                null,
                msg.CorrelationId);
        }
    }

    private async Task HandleRestartDriverAsync(RestartDriver msg)
    {
        var replyTo = Sender;
        try
        {
            // Broadcast to every DriverHostActor on every node via the driver-control DPS topic.
            // Only the host that owns the instance will act; others ignore it (id not found in _children).
            Dispatch(DriverControlTopic.Name, msg);

            await using var db = await _dbFactory.CreateDbContextAsync();
            db.ConfigEdits.Add(new ConfigEdit
            {
                EntityType = "DriverInstance",
                EntityId = Guid.TryParse(msg.DriverInstanceId, out var guid) ? guid : Guid.Empty,
                FieldsJson = $"{{\"op\":\"restart\",\"driverInstanceId\":{System.Text.Json.JsonSerializer.Serialize(msg.DriverInstanceId)}}}",
                EditedBy = msg.ActorByUserName,
                SourceNode = Akka.Cluster.Cluster.Get(Context.System).SelfAddress.Host ?? "unknown",
            });
            await db.SaveChangesAsync();

            _log.Info("AdminOps: RestartDriver dispatched for {DriverInstanceId} by {User}",
                msg.DriverInstanceId, msg.ActorByUserName);
            replyTo.Tell(new RestartDriverResult(true, null, msg.CorrelationId));
        }
        catch (Exception ex)
        {
            _log.Error(ex, "AdminOps: RestartDriver failed for {DriverInstanceId} (correlation {CorrelationId})", msg.DriverInstanceId, msg.CorrelationId);
            replyTo.Tell(new RestartDriverResult(false, ServerFault("RestartDriver", msg.CorrelationId), msg.CorrelationId));
        }
    }

    private async Task HandleReconnectDriverAsync(ReconnectDriver msg)
    {
        var replyTo = Sender;
        try
        {
            // Broadcast to every DriverHostActor; only the one owning the instance reacts.
            Dispatch(DriverControlTopic.Name, msg);

            await using var db = await _dbFactory.CreateDbContextAsync();
            db.ConfigEdits.Add(new ConfigEdit
            {
                EntityType = "DriverInstance",
                EntityId = Guid.TryParse(msg.DriverInstanceId, out var guid) ? guid : Guid.Empty,
                FieldsJson = $"{{\"op\":\"reconnect\",\"driverInstanceId\":{System.Text.Json.JsonSerializer.Serialize(msg.DriverInstanceId)}}}",
                EditedBy = msg.ActorByUserName,
                SourceNode = Akka.Cluster.Cluster.Get(Context.System).SelfAddress.Host ?? "unknown",
            });
            await db.SaveChangesAsync();

            _log.Info("AdminOps: ReconnectDriver dispatched for {DriverInstanceId} by {User}",
                msg.DriverInstanceId, msg.ActorByUserName);
            replyTo.Tell(new ReconnectDriverResult(true, null, msg.CorrelationId));
        }
        catch (Exception ex)
        {
            _log.Error(ex, "AdminOps: ReconnectDriver failed for {DriverInstanceId} (correlation {CorrelationId})", msg.DriverInstanceId, msg.CorrelationId);
            replyTo.Tell(new ReconnectDriverResult(false, ServerFault("ReconnectDriver", msg.CorrelationId), msg.CorrelationId));
        }
    }
}