using System.Collections.Concurrent;
using System.Diagnostics;
using System.IO.Pipes;
using System.Security.Cryptography;
using System.Text;
using Serilog;
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>How a worker process ended.</summary>
internal enum WorkerExitReason
{
    /// <summary>It exited by itself with code 0 (a Shutdown).</summary>
    Clean,

    /// <summary>The host killed it: a script crossed its hard deadline (<see cref="WorkerExit.CulpritKey"/>).</summary>
    KilledDeadline,

    /// <summary>The host killed it: no frame for three heartbeat intervals.</summary>
    KilledHeartbeat,

    /// <summary>
    ///     The host killed it: it kept heartbeating but answered nothing — no result, no start, no soft-deadline notice —
    ///     for the progress ceiling while evaluations were outstanding (#827 review F3). Attributed like a heartbeat kill.
    /// </summary>
    KilledUnresponsive,

    /// <summary>The host killed it: a compile did not finish in time (<see cref="WorkerExit.CulpritKey"/>).</summary>
    KilledLoadTimeout,

    /// <summary>It died by itself (OOM, heap limit, fail-fast, external kill).</summary>
    Crash,

    /// <summary>A framing or handshake violation.</summary>
    ProtocolFault,

    /// <summary>The runner is shutting down.</summary>
    Disposed,

    /// <summary>The host stopped an idle worker (the isolation worker after <c>IsolationIdleTimeout</c>); not a failure.</summary>
    Idle,
}

/// <summary>The end of one worker process.</summary>
/// <param name="Reason">Why it ended.</param>
/// <param name="CulpritKey">The scoped key the host blamed, for a deadline or load kill.</param>
/// <param name="ExitCode">The OS exit code, when known.</param>
/// <param name="InFlightKeys">The scoped keys that had an evaluation outstanding when it ended.</param>
/// <param name="LoadingKeys">The scoped keys whose compile (load) was outstanding when it ended.</param>
internal sealed record WorkerExit(WorkerExitReason Reason, string? CulpritKey, int? ExitCode, IReadOnlyList<string> InFlightKeys,
    IReadOnlyList<string> LoadingKeys)
{
    /// <summary>An exit with no load in flight.</summary>
    public WorkerExit(WorkerExitReason reason, string? culpritKey, int? exitCode, IReadOnlyList<string> inFlightKeys)
        : this(reason, culpritKey, exitCode, inFlightKeys, [])
    {
    }
}

/// <summary>One evaluation of a <see cref="WorkerConnection.SubmitBatch"/>.</summary>
/// <param name="Key">The scoped key.</param>
/// <param name="Request">The request (its id is assigned on submit).</param>
/// <param name="SoftTimeout">T.</param>
/// <param name="HardGrace">G.</param>
/// <param name="OnFinished">Runs once when the worker has answered or ended, before the caller resumes (the slot release).</param>
internal sealed record BatchSubmission(string Key, EvalRequest Request, TimeSpan SoftTimeout, TimeSpan HardGrace, Action? OnFinished);

/// <summary>What the host needs to launch one worker.</summary>
/// <param name="Shard">The shard label (<c>0</c>, <c>1</c>, … or <c>iso</c>).</param>
/// <param name="WorkerPath">The worker entry assembly (<c>.dll</c>, run through the dotnet muxer) or apphost.</param>
/// <param name="HeapLimitBytes">The worker's <c>DOTNET_GCHeapHardLimit</c>.</param>
/// <param name="EvalThreads">The worker scheduler's width (1 for the isolation worker).</param>
/// <param name="HardGrace">The worker watchdog's grace past the soft deadline.</param>
/// <param name="HeartbeatInterval">How often the worker sends a heartbeat.</param>
/// <param name="HandshakeTimeout">How long a launch may take to reach Ready.</param>
/// <param name="PeCacheDirectory">The compiled-PE cache directory handed to the worker; null = no cache.</param>
/// <param name="PeCacheKey">The cache's HMAC key (with <paramref name="PeCacheDirectory"/>).</param>
/// <param name="MarkerDirectory">Where the launch marker for the orphan sweep is written; null = none.</param>
/// <param name="LoadTimeout">
///     How long one compile may run, counted from the worker's <c>LoadStarted</c> for it (not from when the host sent
///     it, lmxopcua#790), before the worker is killed and the key struck; null = 60 s.
/// </param>
/// <param name="CompileThreads">The worker's dedicated compile threads; 0 = its default.</param>
/// <param name="WorkerSid">
///     Windows only: the worker identity's SID (<c>Scripting:Worker:Identity</c>). Set, the worker starts under a restricted
///     token (<see cref="RestrictedWorkerLauncher"/>) and the pipe ACL admits it beside the host; null keeps the host's token.
/// </param>
/// <param name="HeartbeatSilenceLimit">
///     How long the worker may send nothing before it is killed (<see cref="WorkerExitReason.KilledHeartbeat"/>); null =
///     three <paramref name="HeartbeatInterval"/>s. Test seam (with <see cref="WorkerConnection.SilenceLimit"/>).
/// </param>
/// <param name="WorkerSoftClock">
///     Ask the worker to run the soft clock from each run's start (lmxopcua#827; see <see cref="WorkerConnection.WorkerSoftClock"/>).
///     False (tests only) keeps the pre-#827 host clock, which is also what a worker that does not echo it gets.
/// </param>
internal sealed record WorkerLaunchSpec(
    string Shard,
    string WorkerPath,
    long HeapLimitBytes,
    int EvalThreads,
    TimeSpan HardGrace,
    TimeSpan HeartbeatInterval,
    TimeSpan HandshakeTimeout,
    string? PeCacheDirectory = null,
    byte[]? PeCacheKey = null,
    string? MarkerDirectory = null,
    TimeSpan? LoadTimeout = null,
    int CompileThreads = 0,
    string? WorkerSid = null,
    bool WorkerSoftClock = true,
    TimeSpan? HeartbeatSilenceLimit = null)
{
    /// <summary>The default compile timeout.</summary>
    public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60);

    /// <summary><see cref="LoadTimeout"/> or its default.</summary>
    public TimeSpan EffectiveLoadTimeout => LoadTimeout ?? DefaultLoadTimeout;
}

/// <summary>
///     One live ScriptWorker process and its pipe (lmxopcua#780, design §3–§6): launch with a cleared, allow-listed
///     environment, a random pipe that accepts exactly one client, a nonce handshake, a frame reader and a coalescing
///     writer, heartbeat supervision, and the kill. It never restarts itself; <see cref="WorkerShard"/> does that.
/// </summary>
internal sealed class WorkerConnection : IAsyncDisposable
{
    /// <summary>
    ///     Environment variables a worker inherits from the host (design §6): the runtime's own knobs, the temp
    ///     directories and the minimum a Windows process needs. Everything else — notably <c>ServerHistorian__ApiKey</c>,
    ///     ConfigDb connection strings and every other <c>Section__Key</c> override — is dropped.
    /// </summary>
    internal static readonly IReadOnlyList<string> InheritedEnvironment =
    [
        "PATH", "TEMP", "TMP", "TMPDIR", "HOME", "SystemRoot", "SYSTEMROOT", "windir", "SystemDrive", "ComSpec",
        "PATHEXT", "NUMBER_OF_PROCESSORS", "PROCESSOR_ARCHITECTURE", "LOCALAPPDATA",
    ];

    private readonly Process _process;
    private readonly NamedPipeServerStream _pipe;
    private readonly IDisposable? _job;
    private readonly IDisposable? _stdio;
    private readonly string _workDir;
    private readonly ILogger _log;
    private readonly Action<LogBatch> _onLogs;
    private readonly ScriptWorkerFrameWriter _writer;
    private readonly ResultDispatcher _dispatcher;
    private readonly ConcurrentDictionary<ulong, PendingEval> _pending = new();
    private readonly ConcurrentDictionary<ulong, LoadInFlight> _loads = new();
    private readonly ConcurrentDictionary<string, LoadEntry> _loaded = new(StringComparer.Ordinal);
    private readonly TaskCompletionSource<WorkerExit> _exited = new(TaskCreationOptions.RunContinuationsAsynchronously);
    private readonly CancellationTokenSource _stopping = new();
    private readonly TimeSpan _heartbeatInterval;
    private TimeSpan _loadTimeout;
    private readonly Action<WorkerExit> _onExiting;
    private readonly string? _markerPath;
    private TaskCompletionSource<DiagnosticsReply>? _diag;
    private Heartbeat? _lastHeartbeat;
    private long _lastUsedTimestamp;
    private int _finished;
    private (WorkerExitReason Reason, string? Culprit)? _killedFor;
    private long _lastFrameTimestamp;

    /// <summary>
    ///     <see cref="Stopwatch.GetTimestamp"/> of the last evaluation progress — a Results frame of any kind — or of the
    ///     submit that made the worker busy again. See <see cref="ProgressCeiling"/>.
    /// </summary>
    private long _lastProgressTimestamp;
    private long _nextEvalId;
    private long _nextLoadId;
    private int _disposed;

    private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, IDisposable? stdio,
        string workDir, ILogger log, Action<LogBatch> onLogs, TimeSpan heartbeatInterval, TimeSpan loadTimeout,
        Action<WorkerExit> onExiting, string? markerPath, bool workerSoftClock, TimeSpan? heartbeatSilenceLimit)
    {
        WorkerSoftClock = workerSoftClock;
        _stdio = stdio;
        _loadTimeout = loadTimeout;
        _onExiting = onExiting;
        _markerPath = markerPath;
        Shard = shard;
        _process = process;
        _pipe = pipe;
        _job = job;
        _workDir = workDir;
        _log = log;
        _onLogs = onLogs;
        _heartbeatInterval = heartbeatInterval;
        _silenceLimitTicks = (heartbeatSilenceLimit ?? heartbeatInterval * 3).Ticks;
        Pid = process.Id;
        // The handshake used seq 1. A failed write means the pipe is gone: the worker is dead or dying.
        _writer = new ScriptWorkerFrameWriter(pipe, OnWriteFault, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, firstSeq: 1,
            synchronous: ScriptWorkerFrameCodec.UseBlockingIo);
        _dispatcher = new ResultDispatcher($"ScriptWorker[{shard}].Results");
    }

    /// <summary>The shard label.</summary>
    public string Shard { get; }

    /// <summary>The worker's process id.</summary>
    public int Pid { get; }

    /// <summary>
    ///     lmxopcua#827: the worker runs each evaluation's soft clock from the moment an evaluation thread starts it and
    ///     reports the crossing, so the soft deadline no longer counts the pipe or the time a starved worker takes to
    ///     START the run. From the start it is wall-clock time, as in process: a pause during the run still counts. The
    ///     host keeps a transport deadline of its own instead (see
    ///     <see cref="PendingEval"/>). Negotiated: asked for in <c>Hello</c>, true only when the worker echoed it in
    ///     <c>Ready</c>; otherwise the host's clock, started at submit, is the soft deadline (the pre-#827 contract).
    /// </summary>
    public bool WorkerSoftClock { get; }

    /// <summary>Completes when the worker has ended, with why.</summary>
    public Task<WorkerExit> Exited => _exited.Task;

    /// <summary>Whether the worker is still serving.</summary>
    public bool IsAlive => !_exited.Task.IsCompleted && !_stopping.IsCancellationRequested;

    /// <summary><see cref="Stopwatch.GetTimestamp"/> when the handshake completed.</summary>
    public long ReadyTimestamp { get; private set; }

    /// <summary>Evaluations currently outstanding on this worker.</summary>
    public int InFlightCount => _pending.Count;

    /// <summary>
    ///     The compile timeout applied to loads that START from now on (test hook: a test warms the worker's compiler
    ///     under the launch value, then shortens it).
    /// </summary>
    internal TimeSpan LoadTimeout
    {
        get => _loadTimeout;
        set => _loadTimeout = value;
    }

    /// <summary>
    ///     How long the worker may send nothing before it is killed for silence: three heartbeat intervals unless the
    ///     launch spec said otherwise. Test hook: a test with a tight heartbeat launches and warms the worker (its cold
    ///     start JITs Roslyn, which on a loaded box can pause it for longer than three short intervals) under a generous
    ///     limit, then tightens it for the phase it is about.
    /// </summary>
    internal TimeSpan SilenceLimit
    {
        get => TimeSpan.FromTicks(Volatile.Read(ref _silenceLimitTicks));
        set => Volatile.Write(ref _silenceLimitTicks, value.Ticks);
    }

    private long _silenceLimitTicks;

    /// <summary>
    ///     The host's own bound on a worker that heartbeats but makes no evaluation progress (#827 review F3). With the
    ///     worker soft clock the host's per-run timer is only the transport deadline, and the kill G is armed by the
    ///     worker's soft notice — so a worker whose watchdog never sends one (a bug; the design treats the worker as
    ///     untrusted) would hold a key's slot, and its tag at BadResourceUnavailable, forever; a deferred run has no host
    ///     clock at all. So: while evaluations are outstanding and none of them has produced a result, start or soft
    ///     notice for 3 × the largest outstanding T + G plus the silence limit, the worker is killed
    ///     (<see cref="WorkerExitReason.KilledUnresponsive"/>, no culprit: attributed like a heartbeat kill, so no key is
    ///     struck on a shard). A legitimate run never gets near it: by T + G it has answered or been killed. Null = that
    ///     rule; set = an override (tests).
    /// </summary>
    internal TimeSpan? ProgressCeilingOverride { get; set; }

    /// <summary>Test seam: ignore the worker's soft-deadline notices, as a worker whose watchdog never sends them would.</summary>
    internal bool DropSoftNoticesForTest { get; set; }

    /// <summary>The progress ceiling now (see <see cref="ProgressCeilingOverride"/>); null = no outstanding deadline.</summary>
    private TimeSpan? ProgressCeiling()
    {
        if (ProgressCeilingOverride is { } fixedCeiling) return fixedCeiling;
        var largest = TimeSpan.Zero;
        foreach (var (_, pending) in _pending)
            if (pending.HardDeadline > largest) largest = pending.HardDeadline;
        return largest == TimeSpan.Zero ? null : (largest * 3) + SilenceLimit;
    }

    /// <summary>Compiles currently outstanding on this worker.</summary>
    public int LoadingCount => _loads.Count;

    /// <summary>The worker's last heartbeat (its in-flight / queued counts and heap), or null before the first.</summary>
    public Heartbeat? LastHeartbeat => Volatile.Read(ref _lastHeartbeat);

    /// <summary><see cref="Stopwatch.GetTimestamp"/> of the last load or evaluation submitted (or of Ready).</summary>
    public long LastUsedTimestamp => Interlocked.Read(ref _lastUsedTimestamp);

    /// <summary>Launches a worker and completes the handshake.</summary>
    /// <param name="spec">What to launch.</param>
    /// <param name="log">Host diagnostics (stderr lines, exits).</param>
    /// <param name="onLogs">Receives the worker's forwarded <c>ctx.Logger</c> batches.</param>
    /// <param name="onExiting">
    ///     Attribution, called synchronously when the worker has ended and BEFORE any outstanding evaluation or load is
    ///     completed or <see cref="Exited"/> is signalled. A kill strike or suspect mark recorded here is therefore in
    ///     place before any caller can see its evaluation end and trigger the script again (the quarantine race,
    ///     lmxopcua#780).
    /// </param>
    /// <param name="ct">Cancellation.</param>
    /// <returns>A ready connection.</returns>
    public static async Task<WorkerConnection> StartAsync(WorkerLaunchSpec spec, ILogger log, Action<LogBatch> onLogs,
        Action<WorkerExit> onExiting, CancellationToken ct)
    {
        var pipeName = "otopcua-sw-" + Convert.ToHexString(RandomNumberGenerator.GetBytes(12)).ToLowerInvariant();
        var nonce = Convert.ToHexString(RandomNumberGenerator.GetBytes(32));
        var workDir = Path.Combine(Path.GetTempPath(), "otopcua-scriptworker", pipeName);
        Directory.CreateDirectory(workDir);

        // One instance, admitted principals only: the pipe accepts exactly one connection. Windows: an explicit,
        // protected ACL — the host account and, under Scripting:Worker:Identity, the worker identity (design §6).
        // Elsewhere: CurrentUserOnly (a Unix socket owned by this user, 0600).
        if (spec.WorkerSid is not null && !OperatingSystem.IsWindows())
            throw new PlatformNotSupportedException("Scripting:Worker:Identity is only supported on Windows");
        var pipe = OperatingSystem.IsWindows()
            ? WorkerPipeSecurity.CreateServer(pipeName, WorkerPipeSecurity.ForCurrentHost(spec.WorkerSid))
            : new NamedPipeServerStream(pipeName, PipeDirection.InOut, maxNumberOfServerInstances: 1,
                PipeTransmissionMode.Byte, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
        Process? process = null;
        IDisposable? job = null;
        IDisposable? stdio = null;
        string? markerPath = null;
        try
        {
            var psi = BuildStartInfo(spec, pipeName, nonce, workDir);
            var shardLabel = spec.Shard;
            void OnStderr(string? line)
            {
                if (line is { Length: > 0 }) log.Warning("ScriptWorker[{Shard}] {WorkerStderr}", shardLabel, line);
            }
            void OnStdout(string? line)
            {
                if (line is { Length: > 0 }) log.Debug("ScriptWorker[{Shard}] stdout {WorkerStdout}", shardLabel, line);
            }
            // Native and JIT memory backstop above the managed heap limit.
            var jobLimit = spec.HeapLimitBytes + 256L * 1024 * 1024;
            if (spec.WorkerSid is { } workerSid && OperatingSystem.IsWindows())
            {
                // The worker reaches only what its identity may: its working directory is the one place it writes.
                RestrictedWorkerLauncher.GrantWorkDirectory(workDir, workerSid);
                var launched = RestrictedWorkerLauncher.Start(psi, workerSid, jobLimit, OnStdout, OnStderr);
                process = launched.Process;
                job = launched.Job;
                stdio = launched;
                if (spec.MarkerDirectory is { } restrictedMarkerDir)
                    markerPath = WorkerLaunchMarker.Write(restrictedMarkerDir, process, pipeName, log);
            }
            else
            {
                process = new Process { StartInfo = psi, EnableRaisingEvents = true };
                if (!process.Start()) throw new InvalidOperationException("the worker process did not start");
                // Before anything else can fail: a host that dies from here on leaves a marker the next start's orphan
                // sweep can match to exactly this process (design §3.3).
                if (spec.MarkerDirectory is { } markerDir) markerPath = WorkerLaunchMarker.Write(markerDir, process, pipeName, log);
                if (OperatingSystem.IsWindows()) job = WindowsJobObject.Assign(process, jobLimit);
                process.ErrorDataReceived += (_, e) => OnStderr(e.Data);
                process.OutputDataReceived += (_, e) => OnStdout(e.Data);
                process.BeginErrorReadLine();
                process.BeginOutputReadLine();
            }

            using var handshakeCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
            handshakeCts.CancelAfter(spec.HandshakeTimeout);
            var exitedEarly = process.WaitForExitAsync(handshakeCts.Token);
            var accept = pipe.WaitForConnectionAsync(handshakeCts.Token);
            if (await Task.WhenAny(accept, exitedEarly).ConfigureAwait(false) != accept)
                throw new InvalidOperationException(
                    $"the worker exited (code {SafeExitCode(process)}) before connecting to its pipe");
            await accept.ConfigureAwait(false);

            var hello = new Hello
            {
                Nonce = nonce,
                MaxFrameBytes = ScriptWorkerFrameCodec.DefaultMaxFrameBytes,
                EvalThreads = (uint)spec.EvalThreads,
                QueueCapacity = 4096,
                HardGraceMs = (uint)Math.Clamp(spec.HardGrace.TotalMilliseconds, 1, uint.MaxValue),
                HeartbeatIntervalMs = (uint)spec.HeartbeatInterval.TotalMilliseconds,
                MaxLogLines = 1000,
                CompileThreads = (uint)Math.Max(0, spec.CompileThreads),
                WorkerSoftClock = spec.WorkerSoftClock,
            };
            if (spec.PeCacheDirectory is { } cacheDir && spec.PeCacheKey is { } cacheKey)
            {
                hello.PeCacheDir = cacheDir;
                hello.PeCacheKey = Google.Protobuf.ByteString.CopyFrom(cacheKey);
            }
            var ready = ScriptWorkerFrameCodec.UseBlockingIo
                ? await HandshakeBlockingAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false)
                : await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false);

            var connection = new WorkerConnection(spec.Shard, process, pipe, job, stdio, workDir, log, onLogs, spec.HeartbeatInterval,
                spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock,
                heartbeatSilenceLimit: spec.HeartbeatSilenceLimit);
            connection.Begin();
            return connection;
        }
        catch
        {
            try { if (process is { HasExited: false }) process.Kill(entireProcessTree: true); } catch { /* best effort */ }
            job?.Dispose();
            stdio?.Dispose();
            process?.Dispose();
            await pipe.DisposeAsync().ConfigureAwait(false);
            TryDeleteDirectory(workDir);
            WorkerLaunchMarker.TryDelete(markerPath);
            throw;
        }
    }

    /// <summary>
    ///     The host half of the handshake: send <paramref name="hello"/>, require a <c>Ready</c> as the first frame,
    ///     echoing <paramref name="expectedNonce"/> (constant-time compare) from <paramref name="expectedPid"/>.
    /// </summary>
    /// <param name="stream">The accepted pipe.</param>
    /// <param name="hello">The Hello to send.</param>
    /// <param name="expectedNonce">The nonce put in the worker's environment.</param>
    /// <param name="expectedPid">The launched process id; null skips the check.</param>
    /// <param name="ct">Cancellation (the handshake timeout).</param>
    /// <returns>The worker's Ready.</returns>
    /// <exception cref="ScriptWorkerProtocolException">Anything but a matching Ready.</exception>
    internal static async Task<Ready> HandshakeAsync(Stream stream, Hello hello, string expectedNonce, int? expectedPid, CancellationToken ct)
    {
        await ScriptWorkerFrameCodec.WriteAsync(stream, new Envelope
        {
            ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
            Seq = 1,
            Hello = hello,
        }, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, ct).ConfigureAwait(false);
        await stream.FlushAsync(ct).ConfigureAwait(false);

        var reply = await ScriptWorkerFrameCodec.ReadAsync(stream, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, ct).ConfigureAwait(false);
        return CheckReady(reply, expectedNonce, expectedPid);
    }

    /// <summary>
    ///     As <see cref="HandshakeAsync"/> with blocking I/O only (<see cref="ScriptWorkerFrameCodec.UseBlockingIo"/>):
    ///     one asynchronous call would turn the socket under a Unix pipe non-blocking for the connection's whole life.
    ///     The blocking read runs off the caller and is abandoned at <paramref name="ct"/>; the caller then kills the
    ///     worker and disposes the pipe, which ends it.
    /// </summary>
    /// <param name="stream">The accepted pipe.</param>
    /// <param name="hello">The Hello to send.</param>
    /// <param name="expectedNonce">The nonce put in the worker's environment.</param>
    /// <param name="expectedPid">The launched process id; null skips the check.</param>
    /// <param name="ct">Cancellation (the handshake timeout).</param>
    /// <returns>The worker's Ready.</returns>
    internal static async Task<Ready> HandshakeBlockingAsync(Stream stream, Hello hello, string expectedNonce, int? expectedPid,
        CancellationToken ct)
    {
        var reply = await Task.Run(() =>
        {
            ScriptWorkerFrameCodec.Write(stream, new Envelope
            {
                ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
                Seq = 1,
                Hello = hello,
            }, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
            return ScriptWorkerFrameCodec.Read(stream, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
        }, CancellationToken.None).WaitAsync(ct).ConfigureAwait(false);
        return CheckReady(reply, expectedNonce, expectedPid);
    }

    private static Ready CheckReady(Envelope? reply, string expectedNonce, int? expectedPid)
    {
        if (reply is null) throw new ScriptWorkerProtocolException("the worker closed the pipe during the handshake");
        if (reply.BodyCase != Envelope.BodyOneofCase.Ready)
            throw new ScriptWorkerProtocolException($"expected Ready, got {reply.BodyCase}");
        var expected = Encoding.UTF8.GetBytes(expectedNonce);
        var actual = Encoding.UTF8.GetBytes(reply.Ready.Nonce ?? string.Empty);
        if (!CryptographicOperations.FixedTimeEquals(expected, actual))
            throw new ScriptWorkerProtocolException("the worker's nonce does not match the launch nonce");
        if (expectedPid is { } pid && reply.Ready.Pid != pid)
            throw new ScriptWorkerProtocolException($"the pipe client reports pid {reply.Ready.Pid}, not the launched {pid}");
        return reply.Ready;
    }

    internal static ProcessStartInfo BuildStartInfo(WorkerLaunchSpec spec, string pipeName, string nonce, string workDir)
    {
        var psi = new ProcessStartInfo
        {
            UseShellExecute = false,
            CreateNoWindow = true,
            RedirectStandardError = true,
            RedirectStandardOutput = true,
            RedirectStandardInput = false,
            WorkingDirectory = workDir,
        };
        if (spec.WorkerPath.EndsWith(".dll", StringComparison.OrdinalIgnoreCase))
        {
            psi.FileName = ResolveMuxer();
            psi.ArgumentList.Add("exec");
            psi.ArgumentList.Add(spec.WorkerPath);
        }
        else
        {
            psi.FileName = spec.WorkerPath;
        }

        // Cleared, then repopulated from an allow-list (design §6). ProcessStartInfo.Environment starts as a copy of
        // the host's environment; ScadaBridge's sandbox inherits it whole, which this design rejects.
        var inherited = psi.Environment.ToDictionary(kv => kv.Key, kv => kv.Value, StringComparer.Ordinal);
        psi.Environment.Clear();
        foreach (var (name, value) in inherited)
        {
            if (value is null) continue;
            if (InheritedEnvironment.Contains(name, StringComparer.OrdinalIgnoreCase)
                || (name.StartsWith("DOTNET_", StringComparison.OrdinalIgnoreCase)
                    && !name.StartsWith("DOTNET_GCHeap", StringComparison.OrdinalIgnoreCase)))
                psi.Environment[name] = value;
        }
        psi.Environment["DOTNET_GCHeapHardLimit"] = "0x" + spec.HeapLimitBytes.ToString("X", System.Globalization.CultureInfo.InvariantCulture);
        psi.Environment["DOTNET_gcServer"] = "0";
        psi.Environment["DOTNET_CLI_TELEMETRY_OPTOUT"] = "1";
        psi.Environment[ScriptWorkerEnvironment.PipeName] = pipeName;
        psi.Environment[ScriptWorkerEnvironment.Nonce] = nonce;
        psi.Environment[ScriptWorkerEnvironment.ParentPid] = Environment.ProcessId.ToString(System.Globalization.CultureInfo.InvariantCulture);
        psi.Environment[ScriptWorkerEnvironment.Shard] = spec.Shard;
        if (spec.WorkerSid is not null)
        {
            // A restricted worker cannot write the host's temp directory; its own working directory is its temp.
            psi.Environment["TEMP"] = workDir;
            psi.Environment["TMP"] = workDir;
            // No debugger / profiler / diagnostics IPC: a restricted token cannot create the runtime's named objects,
            // and nothing should attach to a sandboxed worker anyway.
            psi.Environment["DOTNET_EnableDiagnostics"] = "0";
        }
        return psi;
    }

    /// <summary>The dotnet muxer that runs a framework-dependent worker <c>.dll</c>.</summary>
    /// <returns>The muxer path, or plain <c>dotnet</c> (resolved on PATH) as a last resort.</returns>
    internal static string ResolveMuxer()
    {
        var hostPath = Environment.GetEnvironmentVariable("DOTNET_HOST_PATH");
        if (!string.IsNullOrEmpty(hostPath) && File.Exists(hostPath)) return hostPath;
        var processPath = Environment.ProcessPath;
        if (processPath is not null
            && string.Equals(Path.GetFileNameWithoutExtension(processPath), "dotnet", StringComparison.OrdinalIgnoreCase))
            return processPath;
        var root = Environment.GetEnvironmentVariable("DOTNET_ROOT");
        if (!string.IsNullOrEmpty(root))
        {
            var candidate = Path.Combine(root, OperatingSystem.IsWindows() ? "dotnet.exe" : "dotnet");
            if (File.Exists(candidate)) return candidate;
        }
        return "dotnet";
    }

    private void Begin()
    {
        ReadyTimestamp = Stopwatch.GetTimestamp();
        Volatile.Write(ref _lastFrameTimestamp, ReadyTimestamp);
        Interlocked.Exchange(ref _lastUsedTimestamp, ReadyTimestamp);
        if (!ScriptWorkerFrameCodec.UseBlockingIo) _ = Task.Run(ReadLoopAsync);
        else new Thread(ReadLoopBlocking) { IsBackground = true, Name = $"ScriptWorker[{Shard}].Reader" }.Start();
        _ = Task.Run(HeartbeatMonitorAsync);
        _ = Task.Run(WatchProcessAsync);
    }

    /// <summary>Ensures <paramref name="key"/> is compiled in this worker at <paramref name="sourceHash"/>.</summary>
    /// <param name="key">The scoped key.</param>
    /// <param name="kind">The context kind.</param>
    /// <param name="source">The source.</param>
    /// <param name="sourceHash">SHA-256 of the source.</param>
    /// <param name="inputPaths">The slot order.</param>
    /// <returns>The entry once the load has an outcome (see <see cref="LoadEntry.Outcome"/>).</returns>
    public Task<LoadEntry> EnsureLoadedAsync(string key, ContextKind kind, string source, byte[] sourceHash, IReadOnlyList<string> inputPaths)
        => EnsureLoadedAsync(key, kind, source, sourceHash, Convert.ToHexString(sourceHash), inputPaths);

    /// <summary>As the other overload, with the hash's hex form precomputed (the runner caches it per key, lmxopcua#789).</summary>
    /// <param name="key">The scoped key.</param>
    /// <param name="kind">The context kind.</param>
    /// <param name="source">The source.</param>
    /// <param name="sourceHash">SHA-256 of the source.</param>
    /// <param name="sourceHashHex">Its upper-case hex form.</param>
    /// <param name="inputPaths">The slot order.</param>
    /// <returns>
    ///     The entry once the load has an outcome: <see cref="LoadEntry.Outcome"/> is the worker's verdict, or null when
    ///     the worker ended first; <see cref="LoadEntry.TimedOut"/> marks the compile this connection killed the worker
    ///     for.
    /// </returns>
    public Task<LoadEntry> EnsureLoadedAsync(string key, ContextKind kind, string source, byte[] sourceHash, string sourceHashHex,
        IReadOnlyList<string> inputPaths)
    {
        if (_loaded.TryGetValue(key, out var existing) && existing.SourceHash == sourceHashHex && existing.SameSlots(inputPaths))
            return existing.Ready;

        var entry = new LoadEntry(sourceHashHex, inputPaths);
        _loaded[key] = entry;
        Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp());
        var loadId = (ulong)Interlocked.Increment(ref _nextLoadId);
        var load = new LoadScripts { LoadId = loadId };
        var script = new ScriptSource { Key = key, Kind = kind, Source = source, SourceSha256 = Google.Protobuf.ByteString.CopyFrom(sourceHash) };
        script.InputPaths.AddRange(inputPaths);
        load.Scripts.Add(script);
        var inFlight = new LoadInFlight(loadId, key, entry, this);
        _loads[loadId] = inFlight;
        if (!Send(new Envelope { Load = load }) && _loads.TryRemove(loadId, out _)) inFlight.Finish(null);
        return entry.Ready;
    }

    /// <summary>Forgets a key's compiled form here and in the worker.</summary>
    /// <param name="keys">The scoped keys.</param>
    public void Unload(IReadOnlyCollection<string> keys)
    {
        var unload = new Unload();
        foreach (var key in keys)
            if (_loaded.TryRemove(key, out _)) unload.Keys.Add(key);
        if (unload.Keys.Count > 0) Send(new Envelope { Unload = unload });
    }

    /// <summary>Submits one evaluation. Its <see cref="PendingEval.Completion"/> yields null when the worker ends first.</summary>
    /// <param name="key">The scoped key.</param>
    /// <param name="request">The request (its id is assigned here).</param>
    /// <returns>The pending evaluation.</returns>
    public PendingEval Submit(string key, EvalRequest request) => Submit(key, request, null, TimeSpan.Zero, null);

    /// <summary>
    ///     Submits one evaluation with its two deadlines (design §5.1) on ONE timer (lmxopcua#789): at
    ///     <paramref name="softTimeout"/> <see cref="PendingEval.Outcome"/> yields <see cref="PendingEval.SoftTimedOut"/>,
    ///     and <paramref name="hardGrace"/> later, if the worker still has not answered, the worker is killed with this
    ///     key as the culprit.
    /// </summary>
    /// <param name="key">The scoped key.</param>
    /// <param name="request">The request (its id is assigned here).</param>
    /// <param name="softTimeout">T; null = no deadlines.</param>
    /// <param name="hardGrace">G.</param>
    /// <param name="onFinished">
    ///     Runs once, synchronously, when the worker has answered or ended — BEFORE <see cref="PendingEval.Outcome"/>
    ///     completes — so a caller that resumes from its await never finds its own run still holding the key's slot.
    /// </param>
    /// <returns>The pending evaluation.</returns>
    public PendingEval Submit(string key, EvalRequest request, TimeSpan? softTimeout, TimeSpan hardGrace, Action? onFinished)
    {
        request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId);
        Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp());
        if (_pending.IsEmpty) Interlocked.Exchange(ref _lastProgressTimestamp, Stopwatch.GetTimestamp());
        var pending = new PendingEval(key, request.EvalId, onFinished, _dispatcher);
        if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline, WorkerSoftClock);
        _pending[request.EvalId] = pending;
        var batch = new EvalBatch();
        batch.Items.Add(request);
        if (!Send(new Envelope { Eval = batch }) || _exited.Task.IsCompleted)
        {
            if (_pending.TryRemove(request.EvalId, out _)) pending.Complete(null);
        }
        return pending;
    }

    /// <summary>
    ///     Submits several evaluations as ONE <see cref="EvalBatch"/> frame (lmxopcua#780 batch submit, design §4.5). Each
    ///     item gets its own deadlines and slot release exactly as through <see cref="Submit(string, EvalRequest, TimeSpan?, TimeSpan, Action?)"/>;
    ///     an item the worker has to queue behind busy evaluation threads has its clock stopped until it starts.
    /// </summary>
    /// <param name="items">The evaluations (their ids are assigned here).</param>
    /// <returns>One pending evaluation per item, in order.</returns>
    public PendingEval[] SubmitBatch(IReadOnlyList<BatchSubmission> items)
    {
        ArgumentNullException.ThrowIfNull(items);
        var pendings = new PendingEval[items.Count];
        var batch = new EvalBatch();
        Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp());
        if (_pending.IsEmpty) Interlocked.Exchange(ref _lastProgressTimestamp, Stopwatch.GetTimestamp());
        for (var i = 0; i < items.Count; i++)
        {
            var item = items[i];
            item.Request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId);
            var pending = new PendingEval(item.Key, item.Request.EvalId, item.OnFinished, _dispatcher);
            pending.Arm(item.SoftTimeout, item.HardGrace, OnHardDeadline, WorkerSoftClock);
            _pending[item.Request.EvalId] = pending;
            batch.Items.Add(item.Request);
            pendings[i] = pending;
        }
        if (items.Count > 0 && (!Send(new Envelope { Eval = batch }) || _exited.Task.IsCompleted))
        {
            foreach (var pending in pendings)
                if (_pending.TryRemove(pending.EvalId, out _)) pending.Complete(null);
        }
        return pendings;
    }

    /// <summary>
    ///     The compiled form of <paramref name="key"/> in this worker when it is loaded at <paramref name="sourceHashHex"/>
    ///     with <paramref name="inputPaths"/> and compiled cleanly — the batch fast path, which never waits for a load.
    /// </summary>
    /// <param name="key">The scoped key.</param>
    /// <param name="sourceHashHex">The source hash (hex).</param>
    /// <param name="inputPaths">The slot order.</param>
    /// <returns>The entry, or null when the key needs a (re)load or its load failed.</returns>
    public LoadEntry? TryGetLoaded(string key, string sourceHashHex, IReadOnlyList<string> inputPaths)
        => _loaded.TryGetValue(key, out var existing) && existing.SourceHash == sourceHashHex && existing.SameSlots(inputPaths)
           && existing.Ready.IsCompleted && existing.Outcome is { Ok: true } && !existing.TimedOut
            ? existing
            : null;

    private void OnHardDeadline(PendingEval pending)
    {
        _log.Warning("Script {ScriptKey} exceeded its hard deadline ({Deadline} ms); killing ScriptWorker[{Shard}] pid {Pid}",
            pending.Key, (long)pending.HardDeadline.TotalMilliseconds, Shard, Pid);
        Kill(WorkerExitReason.KilledDeadline, pending.Key);
    }

    /// <summary>Cancels a run's <c>ctx.CancellationToken</c> in the worker (sent at the soft deadline).</summary>
    /// <param name="pending">The evaluation.</param>
    public void Cancel(PendingEval pending)
    {
        if (!pending.Completion.IsCompleted) Send(new Envelope { Cancel = new Protocol.V1.Cancel { EvalId = pending.EvalId } });
    }

    /// <summary>Asks the worker for its process diagnostics (loaded assemblies, environment names).</summary>
    /// <param name="timeout">How long to wait.</param>
    /// <returns>The reply.</returns>
    public async Task<DiagnosticsReply> DiagnosticsAsync(TimeSpan timeout)
    {
        var tcs = new TaskCompletionSource<DiagnosticsReply>(TaskCreationOptions.RunContinuationsAsynchronously);
        Interlocked.Exchange(ref _diag, tcs);
        Send(new Envelope { DiagRequest = new DiagnosticsRequest() });
        return await tcs.Task.WaitAsync(timeout).ConfigureAwait(false);
    }

    /// <summary>Kills the worker (process tree). The first reason given is the one recorded.</summary>
    /// <param name="reason">Why.</param>
    /// <param name="culpritKey">The scoped key blamed, if any.</param>
    public void Kill(WorkerExitReason reason, string? culpritKey)
    {
        lock (_pending)
        {
            _killedFor ??= (reason, culpritKey);
        }
        try
        {
            // The job holds exactly this one process (ACTIVE_PROCESS = 1), so terminating it is the whole tree, at once.
            if (OperatingSystem.IsWindows() && _job is WindowsJobObject job && job.Terminate()) return;
            if (!_process.HasExited) _process.Kill(entireProcessTree: true);
        }
        catch (InvalidOperationException) { /* already gone */ }
        catch (System.ComponentModel.Win32Exception ex)
        {
            _log.Warning(ex, "ScriptWorker[{Shard}] kill of pid {Pid} failed", Shard, Pid);
        }
    }

    /// <summary>Asks the worker to stop, waits briefly, then kills it.</summary>
    /// <returns>A task that completes when the worker has ended.</returns>
    public ValueTask DisposeAsync() => StopAsync(WorkerExitReason.Disposed);

    /// <summary>As <see cref="DisposeAsync"/>, recording <paramref name="reason"/> as the exit reason.</summary>
    /// <param name="reason">Why the host is stopping it (<see cref="WorkerExitReason.Disposed"/> or <see cref="WorkerExitReason.Idle"/>).</param>
    /// <returns>A task that completes when the worker has ended.</returns>
    public async ValueTask StopAsync(WorkerExitReason reason)
    {
        // Both the shard (on exit) and the runner (on shutdown) dispose a connection; only the first does the work.
        if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
        if (!_exited.Task.IsCompleted)
        {
            lock (_pending) { _killedFor ??= (reason, null); }
            Send(new Envelope { Shutdown = new Shutdown() });
            try
            {
                await _process.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
            }
            catch (TimeoutException)
            {
                Kill(WorkerExitReason.Disposed, null);
            }
            try { await _exited.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); } catch (TimeoutException) { }
        }
        _stopping.Cancel();
        _writer.Complete();
        await _pipe.DisposeAsync().ConfigureAwait(false);
        _job?.Dispose();
        _stdio?.Dispose();
        _process.Dispose();
        TryDeleteDirectory(_workDir);
        WorkerLaunchMarker.TryDelete(_markerPath);
    }

    /// <summary>Test hook: closes the host end of the pipe without touching the process, as a dying host would.</summary>
    internal void SeverPipeForTest() => _pipe.Dispose();

    private bool Send(Envelope envelope) => _writer.TryWrite(envelope);

    private void OnWriteFault(Exception ex)
    {
        if (!_stopping.IsCancellationRequested)
            _log.Debug(ex, "ScriptWorker[{Shard}] pipe write ended", Shard);
        Kill(WorkerExitReason.Crash, null);
    }

    /// <summary>
    ///     The reader, on a dedicated thread with blocking reads of a socket that never turns non-blocking
    ///     (<see cref="ScriptWorkerFrameCodec.UseBlockingIo"/>, lmxopcua#789): a frame wakes this thread straight from
    ///     the kernel, instead of the socket engine's epoll thread queueing a thread-pool work item per read. Not on
    ///     Windows, where a synchronous read of an overlapped pipe is itself completed through the I/O completion port;
    ///     the async loop is the cheaper path there.
    /// </summary>
    private void ReadLoopBlocking()
    {
        WorkerExitReason? fault = null;
        var reader = new ScriptWorkerFrameReader(_pipe);
        try
        {
            while (reader.Read() is { } frame) HandleFrame(frame);
        }
        catch (ScriptWorkerProtocolException ex)
        {
            fault = ProtocolFault(ex);
        }
        catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException
                                       or System.Net.Sockets.SocketException)
        {
        }
        _ = PipeEndedAsync(fault);
    }

    private async Task ReadLoopAsync()
    {
        WorkerExitReason? fault = null;
        var reader = new ScriptWorkerFrameReader(_pipe);
        try
        {
            while (await reader.ReadAsync(_stopping.Token).ConfigureAwait(false) is { } frame) HandleFrame(frame);
        }
        catch (OperationCanceledException) { }
        catch (ScriptWorkerProtocolException ex)
        {
            fault = ProtocolFault(ex);
        }
        catch (Exception ex) when (ex is IOException or ObjectDisposedException)
        {
        }
        await PipeEndedAsync(fault).ConfigureAwait(false);
    }

    private WorkerExitReason ProtocolFault(ScriptWorkerProtocolException ex)
    {
        _log.Warning("ScriptWorker[{Shard}] protocol violation: {Message}; killing pid {Pid}", Shard, ex.Message, Pid);
        return WorkerExitReason.ProtocolFault;
    }

    private void HandleFrame(Envelope frame)
    {
        Volatile.Write(ref _lastFrameTimestamp, Stopwatch.GetTimestamp());
        switch (frame.BodyCase)
        {
            case Envelope.BodyOneofCase.Results:
                Volatile.Write(ref _lastProgressTimestamp, Stopwatch.GetTimestamp());
                if (DropSoftNoticesForTest) frame.Results.SoftExpiredEvalIds.Clear();
                // Worker soft clock (lmxopcua#827): the script itself has run for T. Only a worker that agreed to run
                // the clock may send these. Before the results: the writer coalesces frames, and a notice the worker
                // sent ahead of the same run's result must still answer the caller with its timeout, as in process.
                if (frame.Results.SoftExpiredEvalIds.Count > 0 && !WorkerSoftClock)
                    throw new ScriptWorkerProtocolException("soft-deadline notice from a worker that does not run the soft clock");
                foreach (var id in frame.Results.SoftExpiredEvalIds)
                {
                    if (_pending.TryGetValue(id, out var expired)) expired.SoftExpired();
                    else CheckIssued(id, "soft-deadline notice");
                }
                foreach (var result in frame.Results.Items)
                {
                    // An eval id with no outstanding request is a protocol violation (design §6 "what the host
                    // trusts"), except for one that was already completed by a kill race — ignored then.
                    if (_pending.TryRemove(result.EvalId, out var pending)) pending.Complete(result);
                    else CheckIssued(result.EvalId, "result");
                }
                // Batch submit (lmxopcua#780): a run that waited behind busy evaluation threads has its deadlines
                // restarted when it actually starts, so it is never timed out, struck or killed for a sibling's time.
                // The two notices are unordered with respect to each other (see PendingEval.Start).
                foreach (var id in frame.Results.StartedEvalIds)
                {
                    if (_pending.TryGetValue(id, out var started)) started.Start();
                    else CheckIssued(id, "start notice");
                }
                foreach (var id in frame.Results.DeferredEvalIds)
                {
                    if (_pending.TryGetValue(id, out var deferred)) deferred.Defer();
                    else CheckIssued(id, "deferral notice");
                }
                break;
            case Envelope.BodyOneofCase.LoadStarted:
                // The compile timeout starts now (lmxopcua#790). An unknown id is a load already completed by a
                // kill race (ignored) or one never issued (a protocol violation).
                if (_loads.TryGetValue(frame.LoadStarted.LoadId, out var starting)) starting.Started(_loadTimeout);
                else if (frame.LoadStarted.LoadId == 0 || frame.LoadStarted.LoadId > (ulong)Interlocked.Read(ref _nextLoadId))
                    throw new ScriptWorkerProtocolException($"load start for unknown load id {frame.LoadStarted.LoadId}");
                break;
            case Envelope.BodyOneofCase.LoadResult:
                if (_loads.TryRemove(frame.LoadResult.LoadId, out var load)) load.Finish(frame.LoadResult);
                else throw new ScriptWorkerProtocolException($"load result for unknown load id {frame.LoadResult.LoadId}");
                break;
            case Envelope.BodyOneofCase.Logs:
                try { _onLogs(frame.Logs); }
                catch (Exception ex) { _log.Warning(ex, "ScriptWorker[{Shard}] log forwarding threw", Shard); }
                break;
            case Envelope.BodyOneofCase.Heartbeat:
                Volatile.Write(ref _lastHeartbeat, frame.Heartbeat);
                break;
            case Envelope.BodyOneofCase.DiagReply:
                Volatile.Read(ref _diag)?.TrySetResult(frame.DiagReply);
                break;
            case Envelope.BodyOneofCase.Fault:
                _log.Warning("ScriptWorker[{Shard}] reported a fault: {Fault}", Shard, frame.Fault.Message);
                break;
            default:
                throw new ScriptWorkerProtocolException($"unexpected frame {frame.BodyCase} from the worker");
        }
    }

    /// <summary>An id the host never issued is a protocol violation; one it issued and already completed is a benign race.</summary>
    private void CheckIssued(ulong evalId, string what)
    {
        if (evalId == 0 || evalId > (ulong)Interlocked.Read(ref _nextEvalId))
            throw new ScriptWorkerProtocolException($"{what} for unknown eval id {evalId}");
    }

    private async Task PipeEndedAsync(WorkerExitReason? fault)
    {
        // The pipe is gone: the worker is dead or dying. Make sure, then report.
        try
        {
            if (fault is { } f) Kill(f, null);
            try
            {
                await _process.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false);
            }
            catch (TimeoutException)
            {
                Kill(WorkerExitReason.Crash, null);
                try { await _process.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); } catch (TimeoutException) { }
            }
        }
        catch (InvalidOperationException)
        {
            // The process object was disposed by a concurrent shutdown: the worker is already gone.
        }
        Finish();
    }

    private async Task WatchProcessAsync()
    {
        try { await _process.WaitForExitAsync(_stopping.Token).ConfigureAwait(false); }
        catch (OperationCanceledException) { return; }
        catch (InvalidOperationException) { return; }
        // The reader normally finishes first on EOF; this covers a worker that died while the pipe stayed open.
        await Task.Delay(TimeSpan.FromSeconds(2)).ConfigureAwait(false);
        Finish();
    }

    private async Task HeartbeatMonitorAsync()
    {
        using var timer = new PeriodicTimer(_heartbeatInterval);
        try
        {
            while (await timer.WaitForNextTickAsync(_stopping.Token).ConfigureAwait(false))
            {
                if (_exited.Task.IsCompleted) return;
                var silent = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastFrameTimestamp));
                if (silent > SilenceLimit)
                {
                    _log.Warning("ScriptWorker[{Shard}] pid {Pid} sent nothing for {Silent} ms; killing it", Shard, Pid, (long)silent.TotalMilliseconds);
                    Kill(WorkerExitReason.KilledHeartbeat, null);
                    return;
                }
                if (!_pending.IsEmpty && ProgressCeiling() is { } ceiling)
                {
                    var stuck = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastProgressTimestamp));
                    if (stuck > ceiling)
                    {
                        _log.Warning("ScriptWorker[{Shard}] pid {Pid} has {Count} evaluation(s) outstanding and made no progress for {Stuck} ms (ceiling {Ceiling} ms) while still heartbeating; killing it",
                            Shard, Pid, _pending.Count, (long)stuck.TotalMilliseconds, (long)ceiling.TotalMilliseconds);
                        Kill(WorkerExitReason.KilledUnresponsive, null);
                        return;
                    }
                }
            }
        }
        catch (OperationCanceledException) { }
    }

    private void Finish()
    {
        // The reader (on EOF) and the process watcher both get here; only the first reports.
        if (Interlocked.Exchange(ref _finished, 1) != 0) return;
        // No new frame can be queued from here on, so every Submit / load either fails its Send (and completes itself)
        // or was registered before this point and is completed below.
        _writer.Complete();
        int? code = null;
        try { if (_process.HasExited) code = _process.ExitCode; } catch (InvalidOperationException) { }

        (WorkerExitReason Reason, string? Culprit)? killed;
        lock (_pending) { killed = _killedFor; }
        var reason = killed?.Reason ?? (code == 0 ? WorkerExitReason.Clean : WorkerExitReason.Crash);
        var inFlight = _pending.Values.Select(p => p.Key).Distinct(StringComparer.Ordinal).ToArray();
        var loading = _loads.Values.Select(l => l.Key).Distinct(StringComparer.Ordinal).ToArray();
        var exit = new WorkerExit(reason, killed?.Culprit, code, inFlight, loading);

        // Attribution FIRST, synchronously: a kill strike / quarantine / suspect mark must be in place before any
        // evaluation below completes, because completing it releases the key's single-flight slot and the very next
        // trigger is admitted against whatever the quarantine table says at that instant (lmxopcua#780 race fix).
        try { _onExiting(exit); }
        catch (Exception ex) { _log.Error(ex, "ScriptWorker[{Shard}] exit attribution threw", Shard); }
        _exited.TrySetResult(exit);

        // Every outstanding evaluation and load ends now; null tells the waiter "the worker went away".
        foreach (var id in _pending.Keys.ToArray())
            if (_pending.TryRemove(id, out var pending)) pending.Complete(null);
        foreach (var id in _loads.Keys.ToArray())
            if (_loads.TryRemove(id, out var load)) load.Finish(null);
        Volatile.Read(ref _diag)?.TrySetCanceled();
        // Resumes the callers just completed above, then ends; a later Post runs inline.
        _dispatcher.Stop();
    }

    /// <summary>The compile timer of a started load fired: kill the worker, blaming this key alone (lmxopcua#790).</summary>
    private void OnLoadTimeout(LoadInFlight load)
    {
        if (!_loads.ContainsKey(load.Id)) return;
        load.Entry.TimedOut = true;
        _log.Warning("Script {ScriptKey}: compile in ScriptWorker[{Shard}] has run for {Timeout} s since it started; killing the worker",
            load.Key, Shard, _loadTimeout.TotalSeconds);
        Kill(WorkerExitReason.KilledLoadTimeout, load.Key);
    }

    private static string SafeExitCode(Process process)
    {
        try { return process.HasExited ? process.ExitCode.ToString(System.Globalization.CultureInfo.InvariantCulture) : "?"; }
        catch (InvalidOperationException) { return "?"; }
    }

    private static void TryDeleteDirectory(string dir)
    {
        try { if (Directory.Exists(dir)) Directory.Delete(dir, recursive: true); } catch { /* temp dir; best effort */ }
    }

    /// <summary>One load sent to the worker and not yet answered: its compile timer runs from <c>LoadStarted</c>.</summary>
    private sealed class LoadInFlight(ulong id, string key, LoadEntry entry, WorkerConnection owner)
    {
        private Timer? _timer;
        private int _finished;

        public ulong Id { get; } = id;

        public string Key { get; } = key;

        public LoadEntry Entry { get; } = entry;

        /// <summary>The worker has started compiling it: arm the compile timeout (once).</summary>
        public void Started(TimeSpan timeout)
        {
            if (Volatile.Read(ref _finished) != 0 || _timer is not null) return;
            var timer = new Timer(static s => { var l = (LoadInFlight)s!; l.Owner.OnLoadTimeout(l); }, this, Timeout.Infinite, Timeout.Infinite);
            if (Interlocked.CompareExchange(ref _timer, timer, null) is not null) { timer.Dispose(); return; }
            timer.Change(timeout, Timeout.InfiniteTimeSpan);
            if (Volatile.Read(ref _finished) != 0) timer.Dispose();
        }

        private WorkerConnection Owner { get; } = owner;

        /// <summary>The worker answered (<paramref name="result"/>) or ended (null).</summary>
        public void Finish(LoadResult? result)
        {
            if (Interlocked.Exchange(ref _finished, 1) != 0) return;
            Volatile.Read(ref _timer)?.Dispose();
            if (result is null)
            {
                Owner._loaded.TryRemove(new KeyValuePair<string, LoadEntry>(Key, Entry));
                Entry.Complete(null);
                return;
            }
            var outcome = result.Outcomes.FirstOrDefault(o => o.Key == Key)
                          ?? new LoadOutcome { Key = Key, Ok = false, Error = new ScriptError { Kind = ScriptError.Types.Kind.Threw, Message = "no load outcome" } };
            if (!outcome.Ok) Owner._loaded.TryRemove(new KeyValuePair<string, LoadEntry>(Key, Entry));
            Entry.Complete(outcome);
        }
    }

    /// <summary>A key's compiled form in this worker.</summary>
    internal sealed class LoadEntry
    {
        private readonly TaskCompletionSource<LoadEntry> _ready = new(TaskCreationOptions.RunContinuationsAsynchronously);

        /// <summary>Set once the runner has counted this load in its metrics.</summary>
        public int Counted;

        public LoadEntry(string sourceHash, IReadOnlyList<string> inputPaths)
        {
            SourceHash = sourceHash;
            InputPaths = inputPaths;
            var slots = new Dictionary<string, uint>(inputPaths.Count, StringComparer.Ordinal);
            for (var i = 0; i < inputPaths.Count; i++) slots[inputPaths[i]] = (uint)i;
            Slots = slots;
        }

        public string SourceHash { get; }

        public IReadOnlyList<string> InputPaths { get; }

        /// <summary>Path → slot in the registered order.</summary>
        public IReadOnlyDictionary<string, uint> Slots { get; }

        /// <summary>The worker's verdict; null until it arrives, and null for good when the worker ended first.</summary>
        public LoadOutcome? Outcome { get; private set; }

        /// <summary>
        ///     Set when this connection killed the worker because THIS compile overran its timeout from its start
        ///     (lmxopcua#790); the waiters then report a compile timeout instead of a worker restart.
        /// </summary>
        public volatile bool TimedOut;

        /// <summary>Completes with this entry once the load has an outcome (or the worker ended).</summary>
        public Task<LoadEntry> Ready => _ready.Task;

        public bool SameSlots(IReadOnlyList<string> inputPaths)
        {
            if (inputPaths.Count != InputPaths.Count) return false;
            for (var i = 0; i < inputPaths.Count; i++)
                if (!string.Equals(inputPaths[i], InputPaths[i], StringComparison.Ordinal)) return false;
            return true;
        }

        public void Complete(LoadOutcome? outcome)
        {
            Outcome = outcome;
            _ready.TrySetResult(this);
        }
    }
}

/// <summary>
///     One evaluation outstanding on a worker. Its deadlines share ONE timer (lmxopcua#789: the per-eval
///     <c>WaitAsync(timeout)</c> timer, the hard-deadline <c>CancellationTokenSource</c> and their continuations were a
///     measurable share of the host's CPU).
/// </summary>
/// <remarks>
///     <para>
///         <b>Worker soft clock</b> (lmxopcua#827, <see cref="WorkerConnection.WorkerSoftClock"/>): the soft deadline T is
///         the worker's — it reports, through <see cref="SoftExpired"/>, that the script has itself run for T, and the
///         hard deadline G is armed from that notice. The host's timer is then only a TRANSPORT deadline: T + G from
///         submit (restarted at a deferred run's start). Passing it with neither a result nor a soft notice means the
///         pipe or the worker process is late, not the script — <see cref="Outcome"/> yields
///         <see cref="TransportTimedOut"/> (a runner fault: nothing is struck and the worker is not killed; a worker that
///         is really hung goes silent and the heartbeat monitor kills it). So a starved worker can no longer make a
///         trivial script report "timed out", which in Shadow mode read as a script mismatch.
///     </para>
///     <para>
///         <b>Host clock</b> (a worker that did not agree to the above): the timer fires at the soft deadline T, counted
///         from submit, and is re-armed for the grace G.
///     </para>
///     <para>
///         The deadlines are absolute timestamps the timer callback re-checks, so a deferral (the worker queued the run
///         behind busy evaluation threads, lmxopcua#780 batch submit) can stop the clock and a start notice can restart
///         it — the full T (or transport deadline), then G, from the start — without a stale tick firing early.
///     </para>
/// </remarks>
internal sealed class PendingEval
{
    /// <summary>What <see cref="Outcome"/> yields when the soft deadline passed first. Never on the wire.</summary>
    internal static readonly EvalResult SoftTimedOut = new();

    /// <summary>
    ///     What <see cref="Outcome"/> yields when the transport deadline passed with no word from the worker (worker soft
    ///     clock only, lmxopcua#827): a runner fault, not the script's. Never on the wire.
    /// </summary>
    internal static readonly EvalResult TransportTimedOut = new();

    private const int Waiting = 0;
    private const int SoftReported = 1;
    private const int HardFired = 2;
    private const int TransportReported = 3;

    private readonly TaskCompletionSource<EvalResult?> _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);
    private readonly TaskCompletionSource<EvalResult?> _outcome;
    private readonly ResultDispatcher? _dispatcher;
    private readonly Action? _onFinished;
    private readonly object _clock = new();
    private Timer? _timer;
    private Action<PendingEval>? _onHardDeadline;
    private TimeSpan _soft;
    private TimeSpan _grace;
    private bool _workerClock;
    private long _softDue;
    private long _hardDue;
    private long _transportDue;
    private bool _deferred;
    private bool _started;
    private int _phase;
    private int _done;
    private TaskCompletionSource? _lateSoftExpiry;

    public PendingEval(string key, ulong evalId, Action? onFinished = null, ResultDispatcher? dispatcher = null)
    {
        Key = key;
        EvalId = evalId;
        _onFinished = onFinished;
        _dispatcher = dispatcher;
        // With a dispatcher the caller resumes ON the dispatcher's thread (synchronous continuation, no pool hop);
        // without one, on the thread pool as usual.
        _outcome = new TaskCompletionSource<EvalResult?>(dispatcher is null
            ? TaskCreationOptions.RunContinuationsAsynchronously
            : TaskCreationOptions.None);
    }

    /// <summary>The scoped key.</summary>
    public string Key { get; }

    /// <summary>The eval id on the wire.</summary>
    public ulong EvalId { get; }

    /// <summary>T + G, for the kill log line.</summary>
    public TimeSpan HardDeadline { get; private set; }

    /// <summary>Whether the worker reported this run as queued and has not reported it started (tests / diagnostics).</summary>
    internal bool IsDeferred
    {
        get { lock (_clock) return _deferred; }
    }

    /// <summary>The worker's result, or null when the worker ended first.</summary>
    public Task<EvalResult?> Completion => _completion.Task;

    /// <summary>
    ///     After <see cref="TransportTimedOut"/> (null before): completes when the worker then reports the script itself
    ///     over T (slow AND late: it still gets its Cancel and counts as a timeout, #827 review F2), and is cancelled
    ///     when the worker answers or ends first. Continuations run asynchronously, never on the reader.
    /// </summary>
    internal Task? LateSoftExpiry
    {
        get { lock (_clock) return _lateSoftExpiry?.Task; }
    }

    /// <summary>
    ///     What the caller waits on: the worker's result, null when the worker ended first, <see cref="SoftTimedOut"/> at
    ///     the soft deadline, or <see cref="TransportTimedOut"/> at the transport deadline (the run itself is still
    ///     outstanding in both of the last two cases).
    /// </summary>
    public Task<EvalResult?> Outcome => _outcome.Task;

    /// <summary>Starts the deadline timer. Called once, before the request is sent.</summary>
    /// <param name="soft">T.</param>
    /// <param name="grace">G.</param>
    /// <param name="onHardDeadline">The kill.</param>
    /// <param name="workerClock">The worker runs the soft clock (lmxopcua#827): this timer is the transport deadline.</param>
    internal void Arm(TimeSpan soft, TimeSpan grace, Action<PendingEval> onHardDeadline, bool workerClock = false)
    {
        _soft = soft;
        _grace = grace;
        _onHardDeadline = onHardDeadline;
        _workerClock = workerClock;
        HardDeadline = soft + grace;
        var now = Stopwatch.GetTimestamp();
        _softDue = now + Ticks(soft);
        _hardDue = _softDue + Ticks(grace);
        _transportDue = _hardDue;
        // Created disarmed and assigned before it can fire, so the soft tick can always re-arm it for the grace.
        _timer = new Timer(static s => ((PendingEval)s!).OnTimer(), this, Timeout.Infinite, Timeout.Infinite);
        _timer.Change(workerClock ? soft + grace : soft, Timeout.InfiniteTimeSpan);
    }

    /// <summary>
    ///     The worker queued this run behind busy evaluation threads: stop the clock until it starts (it is not running,
    ///     so it cannot overrun). Ignored once a start notice has arrived — the two are unordered on the wire.
    /// </summary>
    internal void Defer()
    {
        lock (_clock)
        {
            if (_timer is null || _started || _deferred || Volatile.Read(ref _done) != 0) return;
            _deferred = true;
            TryChange(Timeout.InfiniteTimeSpan);
        }
    }

    /// <summary>
    ///     The worker started this run: a deferred run's deadlines restart from now (T, or T + G when its soft deadline
    ///     had already been reported; with the worker soft clock, the transport deadline T + G). A run that was never
    ///     deferred keeps the clock it has had since submit.
    /// </summary>
    internal void Start()
    {
        lock (_clock)
        {
            _started = true;
            if (_timer is null || !_deferred || Volatile.Read(ref _done) != 0) return;
            _deferred = false;
            var now = Stopwatch.GetTimestamp();
            if (_workerClock)
            {
                // Its soft deadline is the worker's from here; the transport deadline restarts. Already answered (a
                // deferral notice that arrived after the transport deadline): nothing to re-arm — a soft notice or the
                // result is what comes next.
                if (Volatile.Read(ref _phase) != Waiting) return;
                _transportDue = now + Ticks(_soft + _grace);
                TryChange(_soft + _grace);
            }
            else if (Volatile.Read(ref _phase) == Waiting)
            {
                _softDue = now + Ticks(_soft);
                _hardDue = _softDue + Ticks(_grace);
                TryChange(_soft);
            }
            else
            {
                // Its caller already has the soft timeout (a deferral notice that arrived after T): the run still gets
                // the whole T + G from its start before it can be blamed for a kill.
                _hardDue = now + Ticks(_soft + _grace);
                TryChange(_soft + _grace);
            }
        }
    }

    /// <summary>
    ///     The worker reports that the script has itself run for T (worker soft clock, lmxopcua#827): the caller gets its
    ///     soft timeout now — unless the transport deadline already answered it — and the hard deadline is G from here.
    /// </summary>
    internal void SoftExpired()
    {
        bool signal, late;
        lock (_clock)
        {
            if (_timer is null || !_workerClock || Volatile.Read(ref _done) != 0) return;
            var phase = Volatile.Read(ref _phase);
            if (phase is SoftReported or HardFired) return;
            _started = true;
            _deferred = false;
            signal = phase == Waiting;
            late = phase == TransportReported;
            Interlocked.Exchange(ref _phase, SoftReported);
            _hardDue = Stopwatch.GetTimestamp() + Ticks(_grace);
            TryChange(_grace);
        }
        // This runs on the pipe READER. The caller's resumption is the caller's code (the seam's publish, the runner's
        // Cancel write): it goes through the dispatcher exactly like a result (ResultDispatcher remarks), never inline
        // here, or a slow consumer would hold up heartbeats and results (#827 review F1).
        if (signal)
        {
            if (_dispatcher is { } dispatcher) dispatcher.Post(this, SoftTimedOut);
            else _outcome.TrySetResult(SoftTimedOut);
        }
        // The caller was already answered "worker unresponsive"; the script has now itself run for T. Its watcher
        // (the runner) cancels the run's token and counts the timeout, off this thread: the completion source runs its
        // continuations asynchronously (#827 review F2).
        if (late) _lateSoftExpiry?.TrySetResult();
    }

    private void OnTimer()
    {
        if (Volatile.Read(ref _done) != 0) return;
        bool soft = false, hard = false, transport = false;
        lock (_clock)
        {
            if (_deferred || Volatile.Read(ref _done) != 0) return;
            var now = Stopwatch.GetTimestamp();
            var phase = Volatile.Read(ref _phase);
            if (phase == Waiting && _workerClock)
            {
                // A tick armed before a deferral/start can land early: re-arm for what is left.
                if (now < _transportDue) { TryChange(Stopwatch.GetElapsedTime(now, _transportDue)); return; }
                Interlocked.Exchange(ref _phase, TransportReported);
                _lateSoftExpiry = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
                transport = true;
            }
            else if (phase == Waiting)
            {
                // A tick armed before a deferral/start can land early: re-arm for what is left.
                if (now < _softDue) { TryChange(Stopwatch.GetElapsedTime(now, _softDue)); return; }
                Interlocked.Exchange(ref _phase, SoftReported);
                soft = true;
                TryChange(_hardDue > now ? Stopwatch.GetElapsedTime(now, _hardDue) : TimeSpan.Zero);
            }
            else if (phase == SoftReported)
            {
                if (now < _hardDue) { TryChange(Stopwatch.GetElapsedTime(now, _hardDue)); return; }
                Interlocked.Exchange(ref _phase, HardFired);
                hard = true;
            }
        }
        // Outside the lock: a caller with a dispatcher resumes synchronously inside TrySetResult.
        if (soft) _outcome.TrySetResult(SoftTimedOut);
        if (transport) _outcome.TrySetResult(TransportTimedOut);
        if (hard && Volatile.Read(ref _done) == 0) _onHardDeadline?.Invoke(this);
    }

    private void TryChange(TimeSpan due)
    {
        try { _timer?.Change(due < TimeSpan.Zero ? TimeSpan.Zero : due, Timeout.InfiniteTimeSpan); }
        catch (ObjectDisposedException) { /* answered meanwhile */ }
    }

    private static long Ticks(TimeSpan span) => (long)(span.TotalSeconds * Stopwatch.Frequency);

    /// <summary>
    ///     The worker answered (or ended, with null): cancel the deadlines and release the key's slot HERE, on the
    ///     reader, then hand the caller's resumption to the dispatcher (or the pool).
    /// </summary>
    public void Complete(EvalResult? result)
    {
        if (Interlocked.Exchange(ref _done, 1) != 0) return;
        _timer?.Dispose();
        lock (_clock) _lateSoftExpiry?.TrySetCanceled();
        try { _onFinished?.Invoke(); }
        finally
        {
            _completion.TrySetResult(result);
            if (_dispatcher is { } dispatcher) dispatcher.Post(this, result);
            else _outcome.TrySetResult(result);
        }
    }

    /// <summary>Resumes the caller (on the dispatcher's thread). A soft timeout may already have done so.</summary>
    internal void SignalOutcome(EvalResult? result) => _outcome.TrySetResult(result);
}