using System.Collections.Concurrent;
using System.Diagnostics;
using System.Diagnostics.Metrics;
using System.Security.Cryptography;
using System.Text;
using Serilog;
using Serilog.Events;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;

namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;

/// <summary>The out-of-process runner's health (the <c>otopcua.script.runner.state</c> gauge: 0 / 1 / 2).</summary>
public enum ScriptRunnerState
{
    /// <summary>Every shard that has been needed is up.</summary>
    Ready = 0,

    /// <summary>At least one shard (or the isolation worker) is down after a failure; the rest serve.</summary>
    Degraded = 1,

    /// <summary>The crash-loop circuit is open: every shard has been down for <c>CircuitOpenAfter</c> (design §5.3.6).</summary>
    Unavailable = 2,
}

/// <summary>A point-in-time view of the runner (the readiness check and diagnostics).</summary>
/// <param name="State">The overall state.</param>
/// <param name="Shards">Every shard, the isolation worker last.</param>
/// <param name="SuspectCount">Keys currently routed to the isolation worker.</param>
/// <param name="QuarantinedCount">Keys currently quarantined (all seams).</param>
/// <param name="PeCacheEnabled">Whether the compiled-PE cache is in use.</param>
public sealed record ScriptRunnerStatus(
    ScriptRunnerState State,
    IReadOnlyList<ScriptWorkerShardStatus> Shards,
    int SuspectCount,
    int QuarantinedCount,
    bool PeCacheEnabled);

/// <summary>
///     The pre-emptible <see cref="IScriptRunner"/> (lmxopcua#780, design §3–§10): compiled scripts run in a pool of
///     <c>ScriptWorker</c> processes — K shards placed by a stable hash of the script key, plus an isolation worker
///     for suspects — so a script that overruns its hard deadline is stopped by killing its worker, and an OOM, a
///     native crash or a fail-fast takes down a worker instead of the server.
/// </summary>
/// <remarks>
///     <list type="bullet">
///         <item><b>Two deadlines.</b> At the soft deadline <c>T</c> the caller gets today's timeout failure (and the
///         run token is cancelled inside the worker). At <c>T + G</c> (<see cref="ScriptWorkerOptions.EffectiveHardGrace"/>)
///         the worker is killed and the key takes a kill strike.</item>
///         <item><b>Attribution</b> runs synchronously in the worker's exit path, before any evaluation it held is
///         completed — so a quarantine earned by a kill is in place before the key's slot is released and the next
///         trigger is admitted. A deadline kill blames exactly one key; the other evaluations in flight on that worker
///         are victims and report <see cref="ScriptRunOutcome.WorkerRestartedReason"/> (no strike). A crash with no
///         deadline crossed makes every key in flight — evaluating or compiling — a suspect, and suspects run (and
///         compile) in the isolation worker one at a time, where a crash names exactly one key; that is also how a
///         script whose compile keeps crashing its worker is found. A suspect returns to its shard after
///         <see cref="ScriptWorkerOptions.IsolationProbationRuns"/> clean runs.</item>
///         <item><b>Quarantine</b> lives here, in the host, so it survives worker restarts: kill strikes (within an
///         hour), consecutive soft timeouts, or one allocation-budget breach. Reasons keep the
///         <see cref="ScriptRunGuard.QuarantinedReasonPrefix"/>, and a source change or <see cref="Reset"/> lifts them.</item>
///         <item><b>Fail closed.</b> When no worker is ready within <see cref="ScriptWorkerOptions.UnavailableGrace"/>
///         the result is <see cref="ScriptRunStatus.Unavailable"/> — never an in-process run — carrying a
///         <see cref="ScriptRunOutcome.Recovery"/> task the seam re-evaluates on.</item>
///         <item><b>PE cache</b> (design §7): with <see cref="ScriptWorkerOptions.PeCacheDir"/> set, workers load
///         HMAC-verified compiled images instead of re-entering Roslyn after a restart.</item>
///     </list>
/// </remarks>
public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable, IDisposable
{
    /// <summary>The worker entry assembly's file name.</summary>
    public const string WorkerAssemblyFileName = "ZB.MOM.WW.OtOpcUa.ScriptWorker.dll";

    /// <summary>The folder under the host's base directory the Host build places the worker in.</summary>
    public const string WorkerFolderName = "script-worker";

    /// <summary>Failure reason when the worker could not take more work.</summary>
    public const string WorkerBusyReason = ScriptRunOutcome.WorkerBusyReason;

    /// <summary>PE cache entries unused for this long are pruned at startup.</summary>
    internal static readonly TimeSpan PeCacheMaxAge = TimeSpan.FromDays(30);

    /// <summary>At most this many PE cache entries are kept.</summary>
    internal const int PeCacheMaxEntries = 50_000;

    private static readonly TimeSpan KillStrikeWindow = TimeSpan.FromHours(1);

    private static readonly ConcurrentDictionary<OutOfProcessScriptRunner, byte> LiveRunners = new();

    private static readonly Counter<long> CompileCounter = OtOpcUaTelemetry.Meter.CreateCounter<long>(
        "otopcua.script.compile", unit: "{compile}",
        description: "Out-of-process script loads (tag: cache = hit | miss | rejected | uncached | negative).");

    private static readonly Histogram<double> RoundTripHistogram = OtOpcUaTelemetry.Meter.CreateHistogram<double>(
        "otopcua.script.ipc.roundtrip", unit: "us",
        description: "Host-measured submit-to-result time of one out-of-process evaluation (tag: kind).");

    private static readonly Histogram<double> EvalDurationHistogram = OtOpcUaTelemetry.Meter.CreateHistogram<double>(
        "otopcua.script.eval.duration", unit: "us",
        description: "Worker-measured run time of one out-of-process evaluation (tag: kind).");

    private static readonly Counter<long> TransportTimeoutCounter = OtOpcUaTelemetry.Meter.CreateCounter<long>(
        "otopcua.script.ipc.timeouts", unit: "{eval}",
        description: "Out-of-process evaluations answered 'script worker unresponsive': no result and no soft-deadline notice from the worker within T + G of submit — a starved or stalled worker, not a slow script (tag: kind; lmxopcua#827).");

    static OutOfProcessScriptRunner()
    {
        // Gauges observe every live runner (one per node in production; tests make several).
        OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.runner.state",
            static () => LiveRunners.Keys.Select(r => new Measurement<int>((int)r.State)),
            unit: "{state}", description: "Out-of-process script runner state: 0 ready, 1 degraded, 2 unavailable (circuit open).");
        OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.up",
            static () => ShardMeasurements(s => s.Up ? 1 : 0),
            unit: "{process}", description: "1 when the shard's ScriptWorker is live (tag: shard).");
        OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.inflight",
            static () => ShardMeasurements(s => s.InFlight),
            unit: "{eval}", description: "Evaluations outstanding on the shard's ScriptWorker (tag: shard).");
        OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.queue_depth",
            static () => ShardMeasurements(s => s.Queued),
            unit: "{eval}", description: "Evaluations queued inside the shard's ScriptWorker, from its heartbeat (tag: shard).");
        OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.worker.heap_bytes",
            static () => LiveRunners.Keys.SelectMany(r => r.AllShards.Select(s => s.Status()))
                .Select(s => new Measurement<long>(s.HeapBytes, new KeyValuePair<string, object?>("shard", s.Label))),
            unit: "By", description: "The shard's ScriptWorker managed heap, from its heartbeat (tag: shard).");
        OtOpcUaTelemetry.Meter.CreateObservableGauge("otopcua.script.isolation.suspects",
            static () => LiveRunners.Keys.Select(r => new Measurement<int>(r.SuspectKeys.Count)),
            unit: "{script}", description: "Scripts currently routed to the isolation worker.");
    }

    private readonly ScriptingOptions _options;
    private readonly ScriptWorkerOptions _worker;
    private readonly ILogger _log;
    private readonly ILogger _scriptRoot;
    private readonly WorkerShard[] _shards;
    private readonly WorkerShard _isolation;
    private readonly ConcurrentDictionary<string, KeyState> _keys = new(StringComparer.Ordinal);
    private readonly ConcurrentDictionary<string, string> _compileFailures = new(StringComparer.Ordinal);
    private readonly Timer _maintenance;
    private readonly string _markerDirectory;
    private readonly string? _peCacheDirectory;
    private readonly byte[]? _peCacheKey;
    private readonly int _compileThreads;
    private long _peCacheHits;
    private long _peCacheMisses;
    private long _peCacheRejected;
    private int _maintaining;
    private int _circuitLogged;
    private int _disposed;

    /// <summary>Initializes the runner. No process starts until the first evaluation.</summary>
    /// <param name="options">The node's <c>Scripting</c> options (sanitized again here).</param>
    /// <param name="log">Host diagnostics: starts, kills, quarantines; null ⇒ Serilog's global logger.</param>
    /// <param name="scriptRoot">Fallback logger for forwarded <c>ctx.Logger</c> lines; null ⇒ <paramref name="log"/>.</param>
    public OutOfProcessScriptRunner(ScriptingOptions options, ILogger? log = null, ILogger? scriptRoot = null)
        : this(options, log, scriptRoot, heartbeatInterval: TimeSpan.FromSeconds(1), loadTimeout: TimeSpan.FromSeconds(60))
    {
    }

    internal OutOfProcessScriptRunner(ScriptingOptions options, ILogger? log, ILogger? scriptRoot,
        TimeSpan heartbeatInterval, TimeSpan loadTimeout, TimeSpan? maintenanceInterval = null, string? markerDirectory = null,
        int compileThreads = 0, int evalThreads = 0)
    {
        _compileThreads = compileThreads;
        ArgumentNullException.ThrowIfNull(options);
        _options = options.Sanitized(out _);
        _worker = _options.Worker;
        _log = log ?? Log.Logger;
        _scriptRoot = scriptRoot ?? _log;
        LoadTimeout = loadTimeout;
        WorkerPath = ResolveWorkerPath(_worker);
        // Fail closed (startup also refuses it, ScriptRunnerSelection.Validate): a configured identity that cannot be
        // applied must never silently fall back to the host's own token.
        if (_worker.ValidateIdentity(OperatingSystem.IsWindows()) is { } identityViolation)
            throw new InvalidOperationException($"Scripting misconfiguration: {identityViolation}");
        WorkerSid = string.IsNullOrWhiteSpace(_worker.Identity)
            ? null
            : ScriptWorkerIdentity.TryResolveSid(_worker.Identity, out var sid, out _) ? sid : null;

        // The PE cache is best effort: a directory or key that cannot be used is a Warning and a cache-less runner,
        // never a startup failure (the path RULE is enforced at startup by ScriptRunnerSelection.Validate).
        if (!string.IsNullOrWhiteSpace(_worker.PeCacheDir))
        {
            try
            {
                _peCacheKey = ScriptPeCacheStore.LoadOrCreateKey(_worker.PeCacheDir);
                _peCacheDirectory = _worker.PeCacheDir;
                var pruned = new ScriptPeCacheStore(_peCacheDirectory, _peCacheKey).Prune(PeCacheMaxAge, PeCacheMaxEntries);
                if (pruned > 0) _log.Information("Script PE cache {Directory}: pruned {Count} stale entr(ies)", _peCacheDirectory, pruned);
            }
            catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
            {
                _peCacheKey = null;
                _peCacheDirectory = null;
                _log.Warning(ex, "Script PE cache {Directory} is unusable; ScriptWorkers will recompile every script on each start",
                    _worker.PeCacheDir);
            }
        }

        // Design §3.3: workers a dead predecessor of THIS host instance left behind, identified by their launch
        // markers — never by process name.
        _markerDirectory = markerDirectory ?? WorkerLaunchMarker.DirectoryFor(_worker.PeCacheDir);
        var swept = WorkerLaunchMarker.Sweep(_markerDirectory, _log);
        if (swept.Count > 0) _log.Warning("Killed {Count} orphaned ScriptWorker process(es) at startup: {Pids}", swept.Count, swept);

        // Design §3.2 [A-4]: P = max(2, cores / K). A positive evalThreads is a test knob (a small scheduler makes the
        // queue-behind-a-busy-thread case deterministic).
        if (evalThreads <= 0) evalThreads = Math.Max(2, Environment.ProcessorCount / _worker.Shards);
        _shards = new WorkerShard[_worker.Shards];
        for (var i = 0; i < _shards.Length; i++)
        {
            var label = i.ToString(System.Globalization.CultureInfo.InvariantCulture);
            _shards[i] = NewShard(label, isolation: false, evalThreads, heartbeatInterval);
        }
        _isolation = NewShard("iso", isolation: true, evalThreads: 1, heartbeatInterval);

        var tick = maintenanceInterval ?? TimeSpan.FromSeconds(1);
        _maintenance = new Timer(_ => _ = MaintainAsync(), null, tick, tick);
        LiveRunners.TryAdd(this, 0);
    }

    /// <summary>The worker entry assembly this runner launches.</summary>
    public string WorkerPath { get; }

    /// <summary>The worker identity's SID (<c>Scripting:Worker:Identity</c>), or null when workers keep the host's token.</summary>
    public string? WorkerSid { get; }

    /// <summary>The overall state (the readiness check, the state gauge).</summary>
    public ScriptRunnerState State => ComputeState(_shards.Select(s => s.DownFor).ToArray(), _isolation.DownFor, _worker.CircuitOpenAfter);

    /// <summary>A snapshot of the runner for diagnostics and health.</summary>
    /// <returns>The status.</returns>
    public ScriptRunnerStatus Status() => new(
        State,
        AllShards.Select(s => s.Status()).ToArray(),
        SuspectKeys.Count,
        _keys.Count(kv => kv.Value.QuarantineReason is not null),
        _peCacheDirectory is not null);

    /// <summary>
    ///     How long a compile in the worker may run before the worker is killed, counted from the moment a compile thread
    ///     starts it — never the time it spent queued behind other compiles (lmxopcua#790).
    /// </summary>
    internal TimeSpan LoadTimeout { get; }

    /// <summary>The shard workers (tests).</summary>
    internal IReadOnlyList<WorkerShard> Shards => _shards;

    /// <summary>The isolation worker (tests).</summary>
    internal WorkerShard Isolation => _isolation;

    /// <summary>The marker directory this runner writes launch markers to and sweeps at startup.</summary>
    internal string MarkerDirectory => _markerDirectory;

    /// <summary>PE cache outcomes reported by this runner's workers (tests / diagnostics).</summary>
    internal (long Hits, long Misses, long Rejected) PeCacheCounts =>
        (Interlocked.Read(ref _peCacheHits), Interlocked.Read(ref _peCacheMisses), Interlocked.Read(ref _peCacheRejected));

    /// <summary>Test hook: runs inside the continuation that releases a key's single-flight slot, just before it does.</summary>
    internal Action<string>? SlotReleasingForTest { get; set; }

    /// <summary>Test hook: runs once per batch frame <see cref="RunBatch"/> submits, with the number of evaluations in it.</summary>
    internal Action<int>? BatchSubmittedForTest { get; set; }

    /// <summary>Scoped keys currently treated as suspects (tests / diagnostics).</summary>
    internal IReadOnlyCollection<string> SuspectKeys =>
        _keys.Where(kv => kv.Value.Suspect).Select(kv => kv.Key).ToArray();

    private IEnumerable<WorkerShard> AllShards => _shards.Append(_isolation);

    /// <summary>The shard a key is placed on when it is not a suspect: a stable FNV-1a hash, so placement survives restarts.</summary>
    /// <param name="scopedKey">The scoped key.</param>
    /// <param name="shardCount">K.</param>
    /// <returns>The shard index.</returns>
    internal static int ShardIndex(string scopedKey, int shardCount)
    {
        var hash = 2166136261u;
        foreach (var b in Encoding.UTF8.GetBytes(scopedKey))
        {
            hash ^= b;
            hash *= 16777619u;
        }
        return (int)(hash % (uint)shardCount);
    }

    /// <summary>
    ///     The state rule: Unavailable when every shard has been down for at least <paramref name="circuitOpenAfter"/>,
    ///     Degraded when any shard or the isolation worker is down, Ready otherwise.
    /// </summary>
    /// <param name="shardsDownFor">Each regular shard's down time (null = up / never needed).</param>
    /// <param name="isolationDownFor">The isolation worker's.</param>
    /// <param name="circuitOpenAfter">The circuit threshold.</param>
    /// <returns>The state.</returns>
    internal static ScriptRunnerState ComputeState(IReadOnlyList<TimeSpan?> shardsDownFor, TimeSpan? isolationDownFor, TimeSpan circuitOpenAfter)
    {
        if (shardsDownFor.Count > 0 && shardsDownFor.All(d => d is { } down && down >= circuitOpenAfter))
            return ScriptRunnerState.Unavailable;
        return shardsDownFor.Any(d => d is not null) || isolationDownFor is not null
            ? ScriptRunnerState.Degraded
            : ScriptRunnerState.Ready;
    }

    /// <summary>The default worker path, or the configured one.</summary>
    /// <param name="worker">The worker options.</param>
    /// <returns>The absolute worker path.</returns>
    /// <remarks>
    ///     With no <see cref="ScriptWorkerOptions.ExecutablePath"/>, the worker is looked up in <c>script-worker/</c>
    ///     under the host's base directory by <see cref="ResolveWorkerEntry"/>: a self-contained worker (the Windows
    ///     installs, which have no .NET runtime) is started through its apphost, a framework-dependent one (docker-dev)
    ///     through the dotnet muxer.
    /// </remarks>
    public static string ResolveWorkerPath(ScriptWorkerOptions worker)
        => string.IsNullOrWhiteSpace(worker.ExecutablePath)
            ? ResolveWorkerEntry(Path.Combine(AppContext.BaseDirectory, WorkerFolderName))
            : Path.GetFullPath(worker.ExecutablePath);

    /// <summary>The worker's apphost file name on this OS.</summary>
    public static string WorkerAppHostFileName
        => OperatingSystem.IsWindows() ? "ZB.MOM.WW.OtOpcUa.ScriptWorker.exe" : "ZB.MOM.WW.OtOpcUa.ScriptWorker";

    /// <summary>
    ///     The entry a worker folder should be launched through: its apphost when the folder holds a
    ///     <b>self-contained</b> worker, otherwise the entry assembly (run through the dotnet muxer).
    /// </summary>
    /// <param name="folder">The worker folder.</param>
    /// <returns>The absolute path of the apphost or of the entry assembly (which may not exist).</returns>
    /// <remarks>
    ///     A framework-dependent build also emits an apphost, but that apphost needs an installed runtime the
    ///     self-contained Windows hosts do not have, and the muxer path is what docker-dev and the tests have always
    ///     used, so the apphost is chosen only for a self-contained worker. Self-contained is read from the worker's
    ///     <c>runtimeconfig.json</c>: the SDK writes <c>includedFrameworks</c> for a self-contained app and
    ///     <c>framework</c>/<c>frameworks</c> for a framework-dependent one.
    /// </remarks>
    public static string ResolveWorkerEntry(string folder)
    {
        var full = Path.GetFullPath(folder);
        var appHost = Path.Combine(full, WorkerAppHostFileName);
        return File.Exists(appHost) && IsSelfContainedWorker(full) ? appHost : Path.Combine(full, WorkerAssemblyFileName);
    }

    /// <summary>Whether the worker in <paramref name="folder"/> was published self-contained.</summary>
    /// <param name="folder">The worker folder.</param>
    /// <returns>True when its <c>runtimeconfig.json</c> lists <c>includedFrameworks</c>.</returns>
    public static bool IsSelfContainedWorker(string folder)
    {
        var config = Path.Combine(folder, Path.GetFileNameWithoutExtension(WorkerAssemblyFileName) + ".runtimeconfig.json");
        if (!File.Exists(config)) return false;
        try
        {
            using var doc = System.Text.Json.JsonDocument.Parse(File.ReadAllText(config),
                new System.Text.Json.JsonDocumentOptions { CommentHandling = System.Text.Json.JsonCommentHandling.Skip, AllowTrailingCommas = true });
            return doc.RootElement.TryGetProperty("runtimeOptions", out var options)
                   && options.TryGetProperty("includedFrameworks", out _);
        }
        catch (Exception ex) when (ex is System.Text.Json.JsonException or IOException or UnauthorizedAccessException)
        {
            return false;
        }
    }

    /// <inheritdoc />
    public async Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
    {
        ArgumentNullException.ThrowIfNull(invocation);
        if (Volatile.Read(ref _disposed) != 0) return ScriptRunOutcome.Unavailable(ScriptRunOutcome.RunnerUnavailableReason);

        var key = InProcessScriptRunner.ScopedKey(invocation.Kind, invocation.Key);
        var state = Admit(key, invocation.Kind, invocation.Source);
        if (state.QuarantineReason is { } quarantined) return ScriptRunOutcome.Failed(quarantined);

        // The source hash and its negative-cache key are per source, so they live on the key's state (computed once per
        // source, not per evaluation, lmxopcua#789).
        if (_compileFailures.TryGetValue(state.FailureKey, out var knownFailure))
        {
            CompileCounter.Add(1, new KeyValuePair<string, object?>("cache", "negative"));
            return ScriptRunOutcome.Failed(knownFailure);
        }

        if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0) return Refused(key, state);

        var handedOff = false;
        SemaphoreSlim? gate = null;
        var shard = state.Suspect ? _isolation : _shards[ShardIndex(key, _shards.Length)];
        WorkerConnection? connection = null;
        try
        {
            state.ScriptLogger = invocation.ScriptLogger;
            if (shard.IsolationGate is { } isolationGate)
            {
                if (!await isolationGate.WaitAsync(Grace(), ct).ConfigureAwait(false))
                    return Unavailable(ScriptRunOutcome.RunnerUnavailableReason, shard, null);
                gate = isolationGate;
            }

            // Fast path: the shard's worker is up (no state computation, no async hop).
            connection = shard.Current ?? await shard.GetReadyAsync(Grace(), ct).ConfigureAwait(false);
            if (connection is null) return Unavailable(ScriptRunOutcome.RunnerUnavailableReason, shard, null);

            var paths = SlotOrder(state, invocation.Inputs);
            var entry = await connection
                .EnsureLoadedAsync(key, WorkerShard.ToWire(invocation.Kind), invocation.Source, state.Sha, state.ShaHex, paths)
                .WaitAsync(ct).ConfigureAwait(false);
            if (entry.TimedOut)
            {
                // A compile that kept a compile thread busy for LoadTimeout after it STARTED is the script's fault (a
                // pathological source): the connection killed the worker and the exit handler struck the key. Time
                // spent queued behind other compiles is not counted (lmxopcua#790).
                return ScriptRunOutcome.Failed($"compile failure: compile did not finish within {connection.LoadTimeout.TotalSeconds:F0} s");
            }
            if (entry.Outcome is not { } outcome) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection);
            NoteLoad(entry);
            if (!outcome.Ok)
            {
                var reason = outcome.Error?.Kind switch
                {
                    ScriptError.Types.Kind.Compile => $"compile error: {outcome.Error.Message}",
                    ScriptError.Types.Kind.Sandbox => $"sandbox violation: {outcome.Error.Message}",
                    _ => $"compile failure: {outcome.Error?.Message}",
                };
                // Remembered until the next deploy boundary, so a broken script never re-enters Roslyn (design §7).
                _compileFailures.TryAdd(state.FailureKey, reason);
                return ScriptRunOutcome.Failed(reason);
            }

            // From here the in-flight slot (and the isolation gate) belong to the evaluation itself: they are released
            // when the worker answers or dies — not when this caller stops waiting at the soft deadline. That keeps a
            // key single-flight between T and T + G, and the isolation worker strictly one-at-a-time. The release runs
            // inside the connection BEFORE the caller's await resumes, so a caller that awaited one run and starts the
            // next is never refused by the run it already has (the §8 harness found 32 of 200 refused when the release
            // was an unordered continuation). On a worker death, attribution (strike, quarantine, suspect) has already
            // run by then.
            var submitted = Stopwatch.GetTimestamp();
            var pending = connection.Submit(key, BuildRequest(key, entry, invocation), invocation.Timeout,
                _worker.EffectiveHardGrace(invocation.Timeout), SlotRelease(key, state, gate));
            handedOff = true;
            return await AwaitSubmittedAsync(key, state, invocation, shard, connection, pending, submitted, ct).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            return ScriptRunOutcome.Failed("evaluation cancelled");
        }
        finally
        {
            if (!handedOff)
            {
                Interlocked.Exchange(ref state.InFlight, 0);
                gate?.Release();
            }
        }

        // Circuit open (design §5.3.6): answer at once instead of making every trigger wait out the grace. A zero grace
        // still kicks a start that is due, so the shards keep retrying at the backoff cap. Only computed when the
        // shard has no live worker (or for the isolation gate).
        TimeSpan Grace() => State == ScriptRunnerState.Unavailable ? TimeSpan.Zero : _worker.UnavailableGrace;
    }

    /// <inheritdoc />
    /// <remarks>
    ///     Every item that can go straight to a live worker — its key admitted, not quarantined, not a suspect, its script
    ///     already loaded there at this source and slot order — is submitted with its siblings for the same worker as ONE
    ///     <c>EvalBatch</c> frame. Anything else (a first evaluation that needs a compile, a worker that is down, a suspect
    ///     bound for the one-at-a-time isolation worker) takes the full <see cref="RunAsync"/> path on its own, so a batch
    ///     never waits for a load or a restart as a whole. Per-item deadlines are armed at submit and, for an item the worker
    ///     has to queue behind busy evaluation threads, restarted when it starts — so an item waiting behind a runaway
    ///     sibling is neither timed out, struck nor blamed for the kill (it is that kill's victim, reported
    ///     <see cref="ScriptRunOutcome.WorkerRestartedReason"/>, design §9).
    /// </remarks>
    public IReadOnlyList<Task<ScriptRunOutcome>> RunBatch(IReadOnlyList<ScriptInvocation> invocations, CancellationToken ct = default)
    {
        ArgumentNullException.ThrowIfNull(invocations);
        var outcomes = new Task<ScriptRunOutcome>[invocations.Count];
        Dictionary<WorkerConnection, List<(int Index, BatchItem Item)>>? groups = null;
        for (var i = 0; i < invocations.Count; i++)
        {
            var invocation = invocations[i];
            ArgumentNullException.ThrowIfNull(invocation);
            switch (TryPrepareBatchItem(invocation, out var item, out var answered))
            {
                case BatchPreparation.Submit:
                    groups ??= [];
                    if (!groups.TryGetValue(item!.Connection, out var group)) groups[item.Connection] = group = [];
                    group.Add((i, item));
                    break;
                case BatchPreparation.Answered:
                    outcomes[i] = Task.FromResult(answered!);
                    break;
                default:
                    outcomes[i] = RunAsync(invocation, ct);
                    break;
            }
        }
        if (groups is null) return outcomes;

        foreach (var (connection, group) in groups)
        {
            var submissions = new BatchSubmission[group.Count];
            for (var j = 0; j < group.Count; j++)
            {
                var item = group[j].Item;
                submissions[j] = new BatchSubmission(item.Key, BuildRequest(item.Key, item.Entry, item.Invocation),
                    item.Invocation.Timeout, _worker.EffectiveHardGrace(item.Invocation.Timeout), SlotRelease(item.Key, item.State, null));
            }
            var submitted = Stopwatch.GetTimestamp();
            BatchSubmittedForTest?.Invoke(submissions.Length);
            var pendings = connection.SubmitBatch(submissions);
            for (var j = 0; j < group.Count; j++)
            {
                var (index, item) = group[j];
                outcomes[index] = AwaitSubmittedAsync(item.Key, item.State, item.Invocation, item.Shard, connection, pendings[j], submitted, ct);
            }
        }
        return outcomes;
    }

    /// <summary>
    ///     The synchronous half of a batch item: everything <see cref="RunAsync"/> decides before it submits, for the case
    ///     where nothing has to be waited for. Takes the key's single-flight slot when it returns
    ///     <see cref="BatchPreparation.Submit"/>; never when it returns anything else.
    /// </summary>
    private BatchPreparation TryPrepareBatchItem(ScriptInvocation invocation, out BatchItem? item, out ScriptRunOutcome? answered)
    {
        item = null;
        answered = null;
        if (Volatile.Read(ref _disposed) != 0) return BatchPreparation.Full;

        var key = InProcessScriptRunner.ScopedKey(invocation.Kind, invocation.Key);
        var state = Admit(key, invocation.Kind, invocation.Source);
        if (state.QuarantineReason is not null || _compileFailures.ContainsKey(state.FailureKey)) return BatchPreparation.Full;

        var shard = state.Suspect ? _isolation : _shards[ShardIndex(key, _shards.Length)];
        if (shard.IsolationGate is not null || shard.Current is not { } connection) return BatchPreparation.Full;
        if (connection.TryGetLoaded(key, state.ShaHex, SlotOrder(state, invocation.Inputs)) is not { } entry) return BatchPreparation.Full;

        if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0)
        {
            answered = Refused(key, state);
            return BatchPreparation.Answered;
        }
        state.ScriptLogger = invocation.ScriptLogger;
        NoteLoad(entry);
        item = new BatchItem(key, state, invocation, shard, connection, entry);
        return BatchPreparation.Submit;
    }

    /// <summary>
    ///     The slot release of one submitted evaluation: from submit on, the key's single-flight slot (and the isolation
    ///     gate) belong to the evaluation itself and are released when the worker answers or dies — not when the caller
    ///     stops waiting at the soft deadline. That keeps a key single-flight between T and T + G, and the isolation
    ///     worker strictly one-at-a-time. The release runs inside the connection BEFORE the caller's await resumes, so a
    ///     caller that awaited one run and starts the next is never refused by the run it already has (the §8 harness found
    ///     32 of 200 refused when the release was an unordered continuation). On a worker death, attribution (strike,
    ///     quarantine, suspect) has already run by then.
    /// </summary>
    private Action SlotRelease(string key, KeyState state, SemaphoreSlim? heldGate)
    {
        var released = 0;
        return () =>
        {
            if (Interlocked.Exchange(ref released, 1) != 0) return;
            SlotReleasingForTest?.Invoke(key);
            Volatile.Write(ref state.StalledRun, null);
            Interlocked.Exchange(ref state.InFlight, 0);
            heldGate?.Release();
        };
    }

    /// <summary>Waits for one submitted evaluation and turns its result into the caller's outcome.</summary>
    private async Task<ScriptRunOutcome> AwaitSubmittedAsync(string key, KeyState state, ScriptInvocation invocation,
        WorkerShard shard, WorkerConnection connection, PendingEval pending, long submitted, CancellationToken ct)
    {
        try
        {
            var result = ct.CanBeCanceled
                ? await pending.Outcome.WaitAsync(ct).ConfigureAwait(false)
                : await pending.Outcome.ConfigureAwait(false);
            if (ReferenceEquals(result, PendingEval.SoftTimedOut))
            {
                // As in process: the run token is cancelled only once the caller has its timeout, so a polling script
                // ends itself (and frees its slot) without the worker being killed.
                connection.Cancel(pending);
                if (ct.IsCancellationRequested) return ScriptRunOutcome.Failed("evaluation cancelled");
                if (NoteTimeout(key, state, invocation.Timeout)) return ScriptRunOutcome.Failed(state.QuarantineReason!);
                return ScriptRunOutcome.Failed($"{ScriptEvalFailureReasons.TimedOutPrefix}{invocation.Timeout.TotalSeconds:F1}s");
            }

            if (ReferenceEquals(result, PendingEval.TransportTimedOut))
            {
                // lmxopcua#827: T + G passed with neither a result nor a soft-deadline notice. The worker runs the soft
                // clock from the run's start, so this is the pipe or a starved / stalled worker — the runner's fault,
                // never the script's: no timeout strike, no Cancel (the run may not even have started), and the slot
                // stays with the run until the worker answers. The seam re-evaluates when it does (or when the worker
                // is replaced, if the heartbeat monitor kills it).
                TransportTimeoutCounter.Add(1, new KeyValuePair<string, object?>("kind", KindTag(invocation.Kind)));
                // Read by a refusal only while it is incomplete, so a write that loses a race with the slot release
                // is harmless.
                Volatile.Write(ref state.StalledRun, pending.Completion);
                if (ct.IsCancellationRequested) return ScriptRunOutcome.Failed("evaluation cancelled");
                return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason) with { Recovery = pending.Completion };
            }

            if (result is null) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection);
            var kindTag = new KeyValuePair<string, object?>("kind", KindTag(invocation.Kind));
            RoundTripHistogram.Record(Stopwatch.GetElapsedTime(submitted).TotalMicroseconds, kindTag);
            EvalDurationHistogram.Record(result.RunMicros, kindTag);
            return Complete(key, state, result, shard, connection);
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            return ScriptRunOutcome.Failed("evaluation cancelled");
        }
    }

    /// <summary>
    ///     The slot order for this evaluation's inputs: the key's previous order when the input set is the same (the
    ///     steady state — no sort, no allocation, lmxopcua#789), otherwise the ordinal sort of the new set.
    /// </summary>
    private static string[] SlotOrder(KeyState state, IReadOnlyDictionary<string, Core.Abstractions.DataValueSnapshot> inputs)
    {
        var cached = state.Paths;
        if (cached is not null && cached.Length == inputs.Count)
        {
            var same = true;
            foreach (var path in cached)
            {
                if (!inputs.ContainsKey(path)) { same = false; break; }
            }
            if (same) return cached;
        }
        var fresh = inputs.Keys.Order(StringComparer.Ordinal).ToArray();
        state.Paths = fresh;
        return fresh;
    }

    /// <inheritdoc />
    public void Reset(ScriptContextKind kind)
    {
        var prefix = InProcessScriptRunner.ScopedKey(kind, string.Empty);
        var dropped = new List<string>();
        foreach (var key in _keys.Keys)
        {
            if (key.StartsWith(prefix, StringComparison.Ordinal) && _keys.TryRemove(key, out _)) dropped.Add(key);
        }
        foreach (var key in _compileFailures.Keys)
        {
            if (key.StartsWith(prefix, StringComparison.Ordinal)) _compileFailures.TryRemove(key, out _);
        }
        foreach (var shard in AllShards)
            shard.Current?.Unload(dropped);
    }

    /// <inheritdoc />
    public IReadOnlyCollection<string> QuarantinedKeys(ScriptContextKind kind)
    {
        var prefix = InProcessScriptRunner.ScopedKey(kind, string.Empty);
        return _keys
            .Where(kv => kv.Value.QuarantineReason is not null && kv.Key.StartsWith(prefix, StringComparison.Ordinal))
            .Select(kv => kv.Key[prefix.Length..])
            .ToArray();
    }

    /// <summary>Stops every worker.</summary>
    /// <returns>A task that completes when they have ended.</returns>
    public async ValueTask DisposeAsync()
    {
        if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
        LiveRunners.TryRemove(this, out _);
        await _maintenance.DisposeAsync().ConfigureAwait(false);
        await Task.WhenAll(AllShards.Select(s => s.DisposeAsync().AsTask())).ConfigureAwait(false);
    }

    /// <summary>Stops every worker (blocking).</summary>
    public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();

    /// <summary>
    ///     The periodic pass: stop the isolation worker once it has been idle for
    ///     <see cref="ScriptWorkerOptions.IsolationIdleTimeout"/>, and log the circuit opening / closing once each.
    /// </summary>
    /// <returns>A task for the pass.</returns>
    internal async Task MaintainAsync()
    {
        if (Volatile.Read(ref _disposed) != 0 || Interlocked.Exchange(ref _maintaining, 1) != 0) return;
        try
        {
            await _isolation.StopIfIdleAsync(_worker.IsolationIdleTimeout).ConfigureAwait(false);

            var open = State == ScriptRunnerState.Unavailable;
            if (open && Interlocked.Exchange(ref _circuitLogged, 1) == 0)
            {
                _log.Error("Script runner circuit OPEN: every ScriptWorker shard has been down for at least {After}; every compiled script on this node reports {Reason} until a worker starts (retrying at the backoff cap)",
                    _worker.CircuitOpenAfter, ScriptRunOutcome.RunnerUnavailableReason);
            }
            else if (!open && Interlocked.Exchange(ref _circuitLogged, 0) == 1)
            {
                _log.Information("Script runner circuit closed: a ScriptWorker shard is serving again");
            }
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "Script runner maintenance pass failed");
        }
        finally
        {
            Interlocked.Exchange(ref _maintaining, 0);
        }
    }

    private WorkerShard NewShard(string label, bool isolation, int evalThreads, TimeSpan heartbeatInterval)
    {
        var spec = new WorkerLaunchSpec(
            Shard: label,
            WorkerPath: WorkerPath,
            HeapLimitBytes: _worker.HeapLimitBytes,
            EvalThreads: evalThreads,
            HardGrace: _worker.HardGrace ?? TimeSpan.FromSeconds(Math.Max(1, _options.RunTimeout.TotalSeconds)),
            HeartbeatInterval: heartbeatInterval,
            HandshakeTimeout: TimeSpan.FromSeconds(15),
            PeCacheDirectory: _peCacheDirectory,
            PeCacheKey: _peCacheKey,
            MarkerDirectory: _markerDirectory,
            LoadTimeout: LoadTimeout,
            CompileThreads: _compileThreads,
            WorkerSid: WorkerSid);
        return new WorkerShard(label, isolation,
            (onExiting, ct) => WorkerConnection.StartAsync(spec, _log, OnLogs, onExiting, ct),
            OnWorkerExit, _log);
    }

    /// <summary>The shard a key's NEXT evaluation goes to (attribution may just have made it a suspect).</summary>
    private WorkerShard RouteFor(string key, KeyState state)
        => state.Suspect ? _isolation : _shards[ShardIndex(key, _shards.Length)];

    private static ScriptRunOutcome Unavailable(string reason, WorkerShard shard, WorkerConnection? failed)
        => ScriptRunOutcome.Unavailable(reason) with { Recovery = shard.WhenRecoveredAsync(failed) };

    /// <summary>
    ///     The answer to an evaluation whose key's previous run still holds its single-flight slot: the usual
    ///     script-attributable refusal — unless that run was answered "worker unresponsive" (lmxopcua#827) and the
    ///     worker still has not answered it, in which case nothing says the script is slow, and the refusal is the
    ///     runner's fault too (re-evaluated when the stalled run completes).
    /// </summary>
    private static ScriptRunOutcome Refused(string key, KeyState state)
    {
        if (Volatile.Read(ref state.StalledRun) is { IsCompleted: false } stalled)
            return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason) with { Recovery = stalled };
        ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair<string, object?>("script", key));
        return ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason);
    }

    private void NoteLoad(WorkerConnection.LoadEntry entry)
    {
        // Counted once per compile, by the caller that waited for it (an already-loaded key is not a load).
        if (Interlocked.Exchange(ref entry.Counted, 1) != 0 || entry.Outcome is not { Ok: true } outcome) return;
        var tag = outcome.Cache switch
        {
            PeCache.Hit => "hit",
            PeCache.Miss => "miss",
            PeCache.Rejected => "rejected",
            _ => "uncached",
        };
        CompileCounter.Add(1, new KeyValuePair<string, object?>("cache", tag));
        switch (outcome.Cache)
        {
            case PeCache.Hit: Interlocked.Increment(ref _peCacheHits); break;
            case PeCache.Miss: Interlocked.Increment(ref _peCacheMisses); break;
            case PeCache.Rejected:
                Interlocked.Increment(ref _peCacheRejected);
                _log.Warning("Script {ScriptKey}: its PE cache entry failed verification and was deleted; recompiled", outcome.Key);
                break;
        }
    }

    private KeyState Admit(string key, ScriptContextKind kind, string source)
    {
        while (true)
        {
            var state = _keys.GetOrAdd(key, static (_, arg) => new KeyState(arg.source, arg.kind), (source, kind));
            if (ReferenceEquals(state.Source, source) || string.Equals(state.Source, source, StringComparison.Ordinal)) return state;
            // The script changed under this key: a clean slate, exactly as the in-process guard does. An evaluation
            // of the old source still in flight releases the OLD state's slot when it ends.
            if (_keys.TryUpdate(key, new KeyState(source, kind), state)) continue;
        }
    }

    private EvalRequest BuildRequest(string key, WorkerConnection.LoadEntry entry, ScriptInvocation invocation)
    {
        var request = new EvalRequest
        {
            Key = key,
            NowUtcTicks = DateTime.UtcNow.Ticks,
            SoftTimeoutMs = (uint)Math.Clamp(invocation.Timeout.TotalMilliseconds, 1, uint.MaxValue),
            HardGraceMs = (uint)Math.Clamp(_worker.EffectiveHardGrace(invocation.Timeout).TotalMilliseconds, 1, uint.MaxValue),
        };
        foreach (var (path, snapshot) in invocation.Inputs)
        {
            if (snapshot is null) continue;
            var input = new Input
            {
                StatusCode = snapshot.StatusCode,
                SourceTsTicks = snapshot.SourceTimestampUtc is { } source ? VariantCodec.ToUtc(source).Ticks : 0,
                ServerTsTicks = VariantCodec.ToUtc(snapshot.ServerTimestampUtc).Ticks,
            };
            if (entry.Slots.TryGetValue(path, out var slot)) input.Slot = slot;
            else input.Path = path;
            if (VariantCodec.TryEncode(snapshot.Value, out var variant, out var unsupported))
            {
                input.Value = variant;
            }
            else
            {
                // Never silently converted (design §14 #11): the worker answers UNSUPPORTED_TYPE for this input.
                input.Unsupported = true;
                input.UnsupportedType = unsupported ?? "?";
            }
            request.Inputs.Add(input);
        }
        return request;
    }

    private ScriptRunOutcome Complete(string key, KeyState state, EvalResult result, WorkerShard shard, WorkerConnection connection)
    {
        // The script returned control (a value or a throw): the soft-timeout streak resets, as in process.
        Interlocked.Exchange(ref state.ConsecutiveTimeouts, 0);

        if (_options.MaxAllocBytesPerRun > 0 && (long)result.AllocatedBytes > _options.MaxAllocBytesPerRun)
        {
            Quarantine(key, state,
                $"one evaluation allocated {result.AllocatedBytes / (1024.0 * 1024.0):F1} MB, over the " +
                $"{_options.MaxAllocBytesPerRun / (1024.0 * 1024.0):F0} MB budget (Scripting:MaxAllocBytesPerRun); " +
                "the script is not evaluated again until its source changes or it is redeployed",
                "allocation");
            return ScriptRunOutcome.Failed(state.QuarantineReason!);
        }

        if (state.Suspect && Interlocked.Increment(ref state.ProbationRuns) >= _worker.IsolationProbationRuns)
        {
            state.Suspect = false;
            _log.Information("Script {ScriptKey} completed {Runs} clean run(s) in the isolation worker; returning it to its shard",
                key, _worker.IsolationProbationRuns);
        }

        switch (result.OutcomeCase)
        {
            case EvalResult.OutcomeOneofCase.Value:
                try
                {
                    return ScriptRunOutcome.Ok(VariantCodec.Decode(result.Value));
                }
                catch (ScriptWorkerProtocolException ex)
                {
                    return ScriptRunOutcome.Failed($"script worker returned an undecodable value: {ex.Message}");
                }
            case EvalResult.OutcomeOneofCase.Predicate:
                return ScriptRunOutcome.Ok(result.Predicate);
            case EvalResult.OutcomeOneofCase.Error:
                return result.Error.Kind switch
                {
                    ScriptError.Types.Kind.Threw => ScriptRunOutcome.Failed($"script threw: {result.Error.Message}"),
                    ScriptError.Types.Kind.UnsupportedType => ScriptRunOutcome.Failed($"unsupported type: {result.Error.Message}"),
                    ScriptError.Types.Kind.Busy => ScriptRunOutcome.Failed(WorkerBusyReason),
                    ScriptError.Types.Kind.NotLoaded => Unavailable(ScriptRunOutcome.WorkerRestartedReason, shard, connection),
                    _ => ScriptRunOutcome.Failed($"script failed: {result.Error.Message}"),
                };
            default:
                return ScriptRunOutcome.Ok(null);
        }
    }

    private bool NoteTimeout(string key, KeyState state, TimeSpan timeout)
    {
        var count = Interlocked.Increment(ref state.ConsecutiveTimeouts);
        if (count < _options.QuarantineAfterConsecutiveTimeouts) return false;
        Quarantine(key, state,
            $"{count} consecutive evaluation(s) timed out (budget {timeout.TotalMilliseconds:F0} ms); " +
            "the script is not evaluated again until its source changes or it is redeployed",
            "timeouts");
        return true;
    }

    /// <summary>
    ///     Attribution for one worker exit. Called synchronously from the worker's exit path BEFORE any evaluation or
    ///     compile it held is completed (see <see cref="WorkerConnection.StartAsync"/>), so its verdict is in place
    ///     before the affected keys' single-flight slots are released.
    /// </summary>
    /// <param name="shard">The shard whose worker ended.</param>
    /// <param name="exit">How it ended.</param>
    internal void OnWorkerExit(WorkerShard shard, WorkerExit exit)
    {
        switch (exit.Reason)
        {
            case WorkerExitReason.KilledDeadline when exit.CulpritKey is { } culprit:
                KillStrike(culprit, "it exceeded its hard deadline and its worker was killed", "killed");
                break;
            case WorkerExitReason.KilledLoadTimeout when exit.CulpritKey is { } culprit:
                KillStrike(culprit, "its compile did not finish and its worker was killed", "killed");
                break;
            case WorkerExitReason.Crash or WorkerExitReason.KilledHeartbeat or WorkerExitReason.ProtocolFault:
                // Compiles count as well as evaluations: a source that crashes Roslyn (or the load) must be found too.
                var implicated = exit.InFlightKeys.Concat(exit.LoadingKeys).Distinct(StringComparer.Ordinal).ToArray();
                if (shard.IsIsolation)
                {
                    // One evaluation (or compile) at a time here: the crash is this key's (design §5.3.3).
                    foreach (var key in implicated)
                        KillStrike(key, exit.InFlightKeys.Contains(key) ? "it crashed the isolation worker" : "its compile crashed the isolation worker",
                            "isolated_crash");
                }
                else if (implicated.Length > 0)
                {
                    foreach (var key in implicated)
                    {
                        if (!_keys.TryGetValue(key, out var state)) continue;
                        state.Suspect = true;
                        Interlocked.Exchange(ref state.ProbationRuns, 0);
                    }
                    _log.Warning("ScriptWorker[{Shard}] ended ({Reason}) with no deadline crossed; {Count} script(s) in flight move to the isolation worker: {Suspects}",
                        shard.Label, exit.Reason, implicated.Length, string.Join(", ", implicated));
                }
                break;
        }
    }

    private void KillStrike(string key, string why, string metricReason)
    {
        if (!_keys.TryGetValue(key, out var state)) return;
        int strikes;
        lock (state.KillStrikes)
        {
            var now = DateTime.UtcNow;
            state.KillStrikes.Add(now);
            state.KillStrikes.RemoveAll(t => now - t > KillStrikeWindow);
            strikes = state.KillStrikes.Count;
        }
        if (strikes >= _worker.QuarantineAfterKills)
        {
            Quarantine(key, state,
                $"{why} ({strikes} time(s) in the last hour); the script is not evaluated again until its source changes or it is redeployed",
                metricReason);
        }
    }

    private void Quarantine(string key, KeyState state, string why, string metricReason)
    {
        var reason = ScriptRunGuard.QuarantinedReasonPrefix + why;
        if (Interlocked.CompareExchange(ref state.QuarantineReason, reason, null) is not null) return;
        ScriptRunGuard.QuarantinedCounter.Add(1,
            new KeyValuePair<string, object?>("script", key),
            new KeyValuePair<string, object?>("reason", metricReason));
        _scriptRoot.ForContext(ScriptLoggerFactory.ScriptNameProperty, key).Error("Script {ScriptKey} quarantined: {Reason}", key, why);
    }

    private void OnLogs(LogBatch batch)
    {
        foreach (var line in batch.Lines)
        {
            var logger = _keys.TryGetValue(line.Key, out var state) && state.ScriptLogger is { } scoped ? scoped : _scriptRoot;
            var level = line.Level <= (uint)LogEventLevel.Fatal ? (LogEventLevel)line.Level : LogEventLevel.Information;
            if (string.IsNullOrEmpty(line.Exception))
                logger.Write(level, "{ScriptMessage:l}", line.Rendered);
            else
                logger.Write(level, "{ScriptMessage:l}{NewLine}{ScriptException:l}", line.Rendered, Environment.NewLine, line.Exception);
        }
        if (batch.Dropped > 0)
            _log.Warning("ScriptWorker dropped {Dropped} script log line(s): a script is logging faster than the pipe drains", batch.Dropped);
    }

    private static string KindTag(ScriptContextKind kind) => kind switch
    {
        ScriptContextKind.VirtualTag => "virtual_tag",
        ScriptContextKind.Calculation => "calculation",
        ScriptContextKind.AlarmPredicate => "alarm",
        _ => "other",
    };

    private static IEnumerable<Measurement<int>> ShardMeasurements(Func<ScriptWorkerShardStatus, int> value)
        => LiveRunners.Keys.SelectMany(r => r.AllShards.Select(s => s.Status()))
            .Select(s => new Measurement<int>(value(s), new KeyValuePair<string, object?>("shard", s.Label)));

    private enum BatchPreparation
    {
        /// <summary>Run it through the full <see cref="RunAsync"/> path on its own.</summary>
        Full,

        /// <summary>Answered without running (refused).</summary>
        Answered,

        /// <summary>Submit it with the batch; its slot is taken.</summary>
        Submit,
    }

    private sealed record BatchItem(string Key, KeyState State, ScriptInvocation Invocation, WorkerShard Shard,
        WorkerConnection Connection, WorkerConnection.LoadEntry Entry);

    private sealed class KeyState(string source, ScriptContextKind kind)
    {
        private byte[]? _sha;
        private string? _shaHex;
        private string? _failureKey;

        public readonly string Source = source;

        /// <summary>SHA-256 of <see cref="Source"/> (computed on first use; a benign race computes the same bytes).</summary>
        public byte[] Sha => _sha ??= SHA256.HashData(Encoding.UTF8.GetBytes(Source));

        /// <summary>Its upper-case hex form.</summary>
        public string ShaHex => _shaHex ??= Convert.ToHexString(Sha);

        /// <summary>The negative compile cache's key for this source (kind-scoped).</summary>
        public string FailureKey => _failureKey ??= InProcessScriptRunner.ScopedKey(kind, ShaHex);

        /// <summary>The slot order last used for this key (see <see cref="SlotOrder"/>).</summary>
        public string[]? Paths;
        public readonly List<DateTime> KillStrikes = [];
        public int InFlight;
        public int ConsecutiveTimeouts;
        public int ProbationRuns;
        public volatile bool Suspect;
        public string? QuarantineReason;
        public ILogger? ScriptLogger;

        /// <summary>
        ///     The completion of this key's in-flight run while it is answered "worker unresponsive" and not yet answered
        ///     by the worker (lmxopcua#827); null otherwise. Cleared by the slot release.
        /// </summary>
        public Task? StalledRun;
    }
}