using Serilog;
using Serilog.Core;
using Serilog.Events;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;

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

/// <summary>
///     <see cref="ScriptRunGuard"/> — the per-script run quarantine shared by the virtual-tag evaluator, the
///     Calculation evaluator and the scripted-alarm engine. A timed-out script is abandoned, not stopped, so
///     before the guard every re-trigger of a runaway script leaked one more thread-pool thread; the guard
///     refuses to start a run while the previous one is still executing, quarantines a script after three
///     consecutive timeouts (a refusal is not a strike), and quarantines a script whose single run allocates past the budget.
/// </summary>
[Trait("Category", "Unit")]
public sealed class ScriptRunGuardTests
{
    /// <summary>
    ///     A CPU-bound script that outlives every timeout in this file but still ENDS (so the orphaned pool
    ///     thread it leaves behind frees itself after the test). <c>System.DateTime</c> is outside the sandbox
    ///     deny-list.
    /// </summary>
    private const string SpinScript =
        """
        var end = System.DateTime.UtcNow.AddSeconds(6);
        while (System.DateTime.UtcNow < end) { }
        return 1;
        """;

    private const string FastScript = "return 42;";

    /// <summary>Outlives the 60 ms budget until the test sets <see cref="FakeScriptContext.Released"/> — a signal, not a wall-clock spin, so load cannot let it finish inside the budget.</summary>
    private const string BriefSpinScript =
        """
        while (!ctx.Released) { }
        return 1;
        """;

    private static readonly TimeSpan ShortTimeout = TimeSpan.FromMilliseconds(60);

    private static ScriptEvaluator<FakeScriptContext, int> Compile(string source) =>
        ScriptEvaluator<FakeScriptContext, int>.Compile(source);

    private static Task<int> Run(ScriptRunGuard guard, string key, string source,
        ScriptEvaluator<FakeScriptContext, int> evaluator, TimeSpan? timeout = null, CancellationToken ct = default, FakeScriptContext? context = null) =>
        guard.RunAsync(key, source, evaluator, context ?? new FakeScriptContext(), timeout ?? TimeSpan.FromSeconds(5), ct);

    [Fact]
    public async Task Completed_run_returns_the_value_and_releases_the_in_flight_slot()
    {
        var guard = new ScriptRunGuard();
        using var fast = Compile(FastScript);

        var value = await Run(guard, "vt-1", FastScript, fast, ct: TestContext.Current.CancellationToken);

        value.ShouldBe(42);
        guard.InFlightCount.ShouldBe(0);
        guard.StartedRunCount.ShouldBe(1);
        guard.IsQuarantined("vt-1", out _).ShouldBeFalse();
    }

    [Fact]
    public async Task A_timed_out_run_throws_ScriptTimeoutException_and_keeps_the_slot_held_by_the_orphan()
    {
        var guard = new ScriptRunGuard();
        using var spin = Compile(SpinScript);

        await Should.ThrowAsync<ScriptTimeoutException>(() =>
            Run(guard, "vt-1", SpinScript, spin, ShortTimeout, TestContext.Current.CancellationToken));

        // The abandoned script is still executing on its pool thread — the guard knows, the caller does not.
        guard.InFlightCount.ShouldBe(1);
        guard.StartedRunCount.ShouldBe(1);
    }

    [Fact]
    public async Task A_run_while_the_previous_orphan_is_still_executing_is_refused_without_starting_another()
    {
        var guard = new ScriptRunGuard();
        using var spin = Compile(SpinScript);
        await Should.ThrowAsync<ScriptTimeoutException>(() =>
            Run(guard, "vt-1", SpinScript, spin, ShortTimeout, TestContext.Current.CancellationToken));

        var ex = await Should.ThrowAsync<ScriptRunRefusedException>(() =>
            Run(guard, "vt-1", SpinScript, spin, ShortTimeout, TestContext.Current.CancellationToken));

        ex.Message.ShouldBe(ScriptRunGuard.PreviousRunStillRunningReason);
        guard.StartedRunCount.ShouldBe(1, "the refusal must not dispatch a second Task.Run");
        guard.InFlightCount.ShouldBe(1);
        guard.RefusedWhileRunningCount.ShouldBe(1);
    }

    [Fact]
    public async Task Three_consecutive_timeouts_quarantine_the_script_and_a_sibling_keeps_evaluating()
    {
        // Stuck-orphan rule off: it is wall-clock based and not what this test pins.
        var guard = new ScriptRunGuard(new ScriptingOptions { StuckOrphanQuarantineFactor = 0 });
        using var spin = Compile(BriefSpinScript);
        using var fast = Compile(FastScript);
        var ct = TestContext.Current.CancellationToken;
        var held = new FakeScriptContext();

        // Three timeouts in a row — each orphan is released (signal) and drained so the next attempt is a real run,
        // not a refusal (a refusal is not a strike; see the next test).
        await Should.ThrowAsync<ScriptTimeoutException>(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct, held));
        await ReleaseAndDrain(guard, held);
        await Should.ThrowAsync<ScriptTimeoutException>(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct, held));
        await ReleaseAndDrain(guard, held);
        var quarantined = await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct, held));
        await ReleaseAndDrain(guard, held);

        quarantined.Message.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix);
        quarantined.Message.ShouldContain("3 consecutive");
        guard.IsQuarantined("bad", out var reason).ShouldBeTrue();
        reason.ShouldNotBeNull();
        guard.QuarantinedCount.ShouldBe(1);

        // A quarantined script is never dispatched again — the 4th attempt refuses before Task.Run.
        await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct, held));
        guard.StartedRunCount.ShouldBe(3);

        // Positive control: a healthy sibling on the SAME guard still runs — the quarantine is per key.
        (await Run(guard, "good", FastScript, fast, ct: ct)).ShouldBe(42);
        guard.IsQuarantined("good", out _).ShouldBeFalse();
        guard.StartedRunCount.ShouldBe(4);
    }

    /// <summary>
    ///     A refusal is NOT a strike (review fix round 1, I1): on a 1 Hz trigger with a 2 s budget a single
    ///     4 s overrun is one timeout followed by two refusals, and counting those would quarantine a script
    ///     that merely ran long once. A refusal is a Bad publish plus its counter, and it leaves the timeout
    ///     streak untouched — neither adding to it nor resetting it.
    /// </summary>
    [Fact]
    public async Task Refusals_neither_count_toward_nor_reset_the_timeout_streak()
    {
        // Stuck-orphan rule off: with it on, a loaded box could age the held orphan past 10x its budget and turn
        // a refusal into a stuck-quarantine, which is a different (wall-clock) rule from the one under test.
        var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 2, StuckOrphanQuarantineFactor = 0 });
        using var spin = Compile(BriefSpinScript);
        var ct = TestContext.Current.CancellationToken;
        var held = new FakeScriptContext();

        // timeout #1, then refusals while the orphan is still running (held until released) — however many, no quarantine.
        await Should.ThrowAsync<ScriptTimeoutException>(() => Run(guard, "k", BriefSpinScript, spin, ShortTimeout, ct, held));
        for (var i = 0; i < 5; i++)
            await Should.ThrowAsync<ScriptRunRefusedException>(() => Run(guard, "k", BriefSpinScript, spin, ShortTimeout, ct, held));
        guard.IsQuarantined("k", out _).ShouldBeFalse("refusals are not strikes");
        guard.RefusedWhileRunningCount.ShouldBe(5);
        await ReleaseAndDrain(guard, held);

        // …and they did not RESET the streak either: the second real timeout reaches the threshold of 2.
        await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "k", BriefSpinScript, spin, ShortTimeout, ct, held));
        guard.IsQuarantined("k", out _).ShouldBeTrue();
        await ReleaseAndDrain(guard, held);
    }

    /// <summary>
    ///     The thread-leak regression, and the stuck-orphan quarantine (fix round 2). Pre-guard, a
    ///     <c>while(true)</c> script on a 1 Hz trigger leaked one pool thread per trigger — 30 orphans after 30
    ///     triggers, trending toward pool starvation. Now the first trigger's orphan is the only one; later
    ///     triggers are refused while it is young, and once it has run longer than
    ///     <see cref="ScriptingOptions.StuckOrphanQuarantineFactor"/> × the timeout (50 × 60 ms here — the
    ///     trigger cadence is scaled down from 1 Hz to keep the test short) the next refusal quarantines it.
    ///     A healthy sibling keeps evaluating throughout (the positive control an absence assertion needs).
    /// </summary>
    /// <remarks>
    ///     The load-bearing assertions are on the guard's own bookkeeping (<see cref="ScriptRunGuard.InFlightCount"/>
    ///     / <see cref="ScriptRunGuard.StartedRunCount"/>): <see cref="ThreadPool.ThreadCount"/> moves with
    ///     every parallel test class on a loaded machine, so it is kept as a coarse secondary check only —
    ///     it must stay far below the 30 the leak would have produced.
    ///     <para>lmxopcua#843 (failed 2/4 full-suite runs at load average 140–280; 2 of the first 4 runs alone under
    ///     28 CPU burners): this used to label a refusal "young" by the TEST's stopwatch, started before the first
    ///     trigger, and demand one inside 600 ms (10 × 60 ms). Under load the first trigger's dispatch, the sibling
    ///     and the cadence ate those 600 ms, so the guard's answers were <c>timeout, refused, refused,
    ///     quarantined…</c> — the guard right, the test's label wrong. Now: the guard is the authority on the
    ///     orphan's age (a <see cref="ScriptRunRefusedException"/> IS the young-orphan answer; behind a stuck one
    ///     it quarantines instead), the stuck factor is 50 (3 s) so the second trigger has ample room to land
    ///     while the orphan is young, the loop runs until the quarantine instead of a fixed 30 triggers, and the
    ///     one wall-clock check left runs the safe way round — the orphan is never older than the test's
    ///     stopwatch, so a quarantine before 3 s on it is early on any box. The runaway is a HELD spin (released
    ///     at the end) so it cannot end on its own and become a second dispatch on a slow box.</para>
    /// </remarks>
    [Fact]
    public async Task Runaway_script_leaves_one_orphan_is_quarantined_once_stuck_and_a_sibling_keeps_evaluating()
    {
        const double stuckFactor = 50;
        var guard = new ScriptRunGuard(new ScriptingOptions { StuckOrphanQuarantineFactor = stuckFactor });
        using var spin = Compile(BriefSpinScript);
        using var fast = Compile(FastScript);
        var ct = TestContext.Current.CancellationToken;
        var held = new FakeScriptContext();
        var threadsBefore = ThreadPool.ThreadCount;
        var stuckAfter = ShortTimeout * stuckFactor; // 3 s

        var healthyResults = new List<int>();
        var outcomes = new List<string>();
        var timeline = new List<string>();
        TimeSpan? quarantinedAt = null;
        var sw = System.Diagnostics.Stopwatch.StartNew();
        try
        {
            // At least 30 triggers (the leak's scale), then until three quarantined answers; capped as a hang guard.
            for (var trigger = 0; trigger < 1_000 && (trigger < 30 || outcomes.Count(o => o == "quarantined") < 3); trigger++)
            {
                string outcome;
                try { await Run(guard, "runaway", BriefSpinScript, spin, ShortTimeout, ct, held); outcome = "value"; }
                catch (ScriptTimeoutException) { outcome = "timeout"; }
                catch (ScriptRunRefusedException) { outcome = "refused"; }
                catch (ScriptQuarantinedException) { outcome = "quarantined"; quarantinedAt ??= sw.Elapsed; }
                outcomes.Add(outcome);
                timeline.Add($"#{trigger} {outcome} @{sw.ElapsedMilliseconds}ms");

                healthyResults.Add(await Run(guard, "healthy", FastScript, fast, ct: ct));
                await Task.Delay(50, ct);
            }

            var triggers = outcomes.Count;
            var diagnostics = $"timeline: {string.Join(", ", timeline)}";
            guard.InFlightCount.ShouldBe(1, "exactly one orphaned run — the first trigger's — may be executing; " + diagnostics);
            guard.StartedRunCount.ShouldBe(1 + triggers, "one dispatch for the runaway script, one per trigger for the healthy sibling; " + diagnostics);
            outcomes[0].ShouldBe("timeout", diagnostics);
            var firstQuarantine = outcomes.IndexOf("quarantined");
            firstQuarantine.ShouldBeGreaterThan(0, "the stuck orphan was never quarantined; " + diagnostics);
            outcomes.Take(firstQuarantine).Skip(1).ShouldAllBe(o => o == "refused", diagnostics);
            outcomes.Take(firstQuarantine).ShouldContain("refused",
                "a young orphan is only refused — a short overrun is not quarantined; " + diagnostics);
            outcomes.Skip(firstQuarantine).ShouldAllBe(o => o == "quarantined", diagnostics);
            quarantinedAt!.Value.ShouldBeGreaterThanOrEqualTo(stuckAfter,
                $"the orphan is never older than the test's stopwatch, so a quarantine before {stuckFactor}x the budget on it is early; " + diagnostics);
            guard.IsQuarantined("runaway", out var reason).ShouldBeTrue("the orphan outlived its stuck threshold");
            reason!.ShouldContain("stuck");
            guard.QuarantinedCount.ShouldBe(1);
            healthyResults.ShouldAllBe(v => v == 42);
            healthyResults.Count.ShouldBe(triggers);

            var threadDelta = ThreadPool.ThreadCount - threadsBefore;
            threadDelta.ShouldBeLessThan(15, $"{triggers} leaked orphans would have grown the pool by ~{triggers} threads");
        }
        finally
        {
            held.Released = true;
        }
        await WaitForIdle(guard);
    }

    /// <summary>
    ///     Positive control for the stuck rule: a 2x overrun (one timeout, refusals while the orphan is young)
    ///     is NOT quarantined, and once the orphan completes the script evaluates normally again.
    /// </summary>
    [Fact]
    public async Task A_2x_overrun_is_not_quarantined_and_resumes_evaluating_once_the_orphan_completes()
    {
        var guard = new ScriptRunGuard();
        const string source =
            """
            var ms = (int)(ctx.GetTag("spinMs").Value ?? 0);
            var end = System.DateTime.UtcNow.AddMilliseconds(ms);
            while (System.DateTime.UtcNow < end) { }
            return 7;
            """;
        using var eval = Compile(source);
        var ct = TestContext.Current.CancellationToken;
        var timeout = TimeSpan.FromMilliseconds(100);

        FakeScriptContext Ctx(int spinMs)
        {
            var c = new FakeScriptContext();
            c.Tags["spinMs"] = new Core.Abstractions.DataValueSnapshot(spinMs, 0u, c.Now, c.Now);
            return c;
        }

        // Warm the JIT so the first measured run's overrun is the script's, not the compiler's.
        (await guard.RunAsync("k", source, eval, Ctx(0), TimeSpan.FromSeconds(5), ct)).ShouldBe(7);

        // 2x overrun: 200 ms against a 100 ms budget — times out, then is refused while it finishes.
        await Should.ThrowAsync<ScriptTimeoutException>(() => guard.RunAsync("k", source, eval, Ctx(200), timeout, ct));
        await Should.ThrowAsync<ScriptRunRefusedException>(() => guard.RunAsync("k", source, eval, Ctx(0), timeout, ct));
        await WaitForIdle(guard);

        guard.IsQuarantined("k", out _).ShouldBeFalse("a 2x overrun is far below the 10x stuck threshold");
        (await guard.RunAsync("k", source, eval, Ctx(0), timeout, ct)).ShouldBe(7, "the script resumes evaluating");
        guard.QuarantinedCount.ShouldBe(0);
    }

    /// <summary>The stuck rule is a knob: with <see cref="ScriptingOptions.StuckOrphanQuarantineFactor"/> ≤ 0 a
    /// stuck orphan is only ever refused.</summary>
    [Fact]
    public async Task StuckOrphanQuarantineFactor_zero_disables_the_stuck_rule()
    {
        var guard = new ScriptRunGuard(new ScriptingOptions { StuckOrphanQuarantineFactor = 0 });
        using var spin = Compile(SpinScript);
        var ct = TestContext.Current.CancellationToken;

        await Should.ThrowAsync<ScriptTimeoutException>(() => Run(guard, "k", SpinScript, spin, ShortTimeout, ct));
        await Task.Delay(ShortTimeout * 12, ct);
        await Should.ThrowAsync<ScriptRunRefusedException>(() => Run(guard, "k", SpinScript, spin, ShortTimeout, ct));
        guard.IsQuarantined("k", out _).ShouldBeFalse();
    }

    [Fact]
    public async Task A_source_change_clears_the_quarantine_and_abandons_the_old_orphan_slot()
    {
        var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 });
        using var spin = Compile(SpinScript);
        using var fast = Compile(FastScript);
        var ct = TestContext.Current.CancellationToken;

        await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "vt", SpinScript, spin, ShortTimeout, ct));
        guard.IsQuarantined("vt", out _).ShouldBeTrue();

        // The operator fixed the script: a different source identity under the same key starts clean. The
        // old orphan is a sunk cost — it keeps running, but it must not block the FIXED script.
        (await Run(guard, "vt", FastScript, fast, ct: ct)).ShouldBe(42);
        guard.IsQuarantined("vt", out _).ShouldBeFalse();
        guard.InFlightCount.ShouldBe(1, "the abandoned orphan is still counted until it finishes");
    }

    [Fact]
    public async Task Reset_and_ResetAll_clear_a_quarantine()
    {
        var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 });
        using var spin = Compile(SpinScript);
        using var fast = Compile(FastScript);
        var ct = TestContext.Current.CancellationToken;

        await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "a", SpinScript, spin, ShortTimeout, ct));
        await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "b", SpinScript, spin, ShortTimeout, ct));
        guard.QuarantinedKeys.ShouldBe(new[] { "a", "b" }, ignoreOrder: true);

        guard.Reset("a");
        guard.IsQuarantined("a", out _).ShouldBeFalse();
        guard.IsQuarantined("b", out _).ShouldBeTrue();

        guard.ResetAll();
        guard.IsQuarantined("b", out _).ShouldBeFalse();
        // The key is usable again (a fresh slot: the orphan from the quarantined run no longer blocks it).
        (await Run(guard, "b", FastScript, fast, ct: ct)).ShouldBe(42);
    }

    [Fact]
    public async Task A_run_that_allocates_past_the_budget_is_quarantined()
    {
        var guard = new ScriptRunGuard(new ScriptingOptions { MaxAllocBytesPerRun = 1024 * 1024 });
        // Writes into every array so the JIT cannot fold `new byte[N].Length` to a constant and drop the
        // allocation as dead code (which is exactly what it does to the naive one-liner).
        const string hungry =
            "long total = 0; for (var i = 0; i < 32; i++) { var a = new byte[256 * 1024]; a[i] = 1; total += a[i] + a.Length; } return (int)total;";
        using var evaluator = Compile(hungry);
        using var fast = Compile(FastScript);
        var ct = TestContext.Current.CancellationToken;

        var ex = await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "hog", hungry, evaluator, ct: ct));
        ex.Message.ShouldContain("allocat");
        guard.IsQuarantined("hog", out var reason).ShouldBeTrue();
        reason.ShouldNotBeNull();
        reason.ShouldContain("allocat");

        // Never dispatched again, and the slot was released (the run DID complete).
        await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "hog", hungry, evaluator, ct: ct));
        guard.StartedRunCount.ShouldBe(1);
        guard.InFlightCount.ShouldBe(0);

        // Positive control: a modest allocation under the budget is fine.
        (await Run(guard, "ok", FastScript, fast, ct: ct)).ShouldBe(42);
    }

    /// <summary>
    ///     lmxopcua#784 — the quarantine Error reaches the main log through <see cref="ScriptLogCompanionSink"/>.
    ///     The guard's logger is a seam's script root logger, which carries no ScriptName, so the mirror read
    ///     "[Script] unknown emitted Error ..." and every quarantine on the node shared one "unknown" rate-limit
    ///     key. The guard now binds the script instance key: the mirror names each script, and two scripts are
    ///     rate-limited apart (with a budget of 1 per window, the second quarantine is still mirrored).
    /// </summary>
    [Fact]
    public async Task The_quarantine_line_mirrored_to_the_main_log_names_the_script_and_is_rate_limited_per_script()
    {
        var mainSink = new CapturingSink();
        var mainLogger = new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(mainSink).CreateLogger();
        var scriptRoot = new LoggerConfiguration()
            .MinimumLevel.Verbose()
            .WriteTo.Sink(new ScriptLogCompanionSink(mainLogger, limiter: new ScriptLogRateLimiter(1, TimeSpan.FromMinutes(5))))
            .CreateLogger();
        var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 }, scriptRoot);
        using var spin = Compile(SpinScript);
        var ct = TestContext.Current.CancellationToken;

        await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "vt-spin-a", SpinScript, spin, ShortTimeout, ct));
        await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "vt-spin-b", SpinScript, spin, ShortTimeout, ct));

        var mirrored = mainSink.Events.Select(e => e.RenderMessage()).ToList();
        mirrored.Count.ShouldBe(2, string.Join(" | ", mirrored));
        mirrored.ShouldAllBe(m => !m.Contains("unknown"));
        mirrored.ShouldContain(m => m.Contains("vt-spin-a") && m.StartsWith("[Script] \"vt-spin-a\" emitted Error"));
        mirrored.ShouldContain(m => m.Contains("vt-spin-b") && m.StartsWith("[Script] \"vt-spin-b\" emitted Error"));
    }

    private sealed class CapturingSink : ILogEventSink
    {
        public List<LogEvent> Events { get; } = [];

        public void Emit(LogEvent logEvent)
        {
            lock (Events) Events.Add(logEvent);
        }
    }

    [Fact]
    public async Task A_disabled_allocation_budget_never_quarantines()
    {
        var guard = new ScriptRunGuard(new ScriptingOptions { MaxAllocBytesPerRun = 0 });
        // Writes into every array so the JIT cannot fold `new byte[N].Length` to a constant and drop the
        // allocation as dead code (which is exactly what it does to the naive one-liner).
        const string hungry =
            "long total = 0; for (var i = 0; i < 32; i++) { var a = new byte[256 * 1024]; a[i] = 1; total += a[i] + a.Length; } return (int)total;";
        using var evaluator = Compile(hungry);

        (await Run(guard, "hog", hungry, evaluator, ct: TestContext.Current.CancellationToken)).ShouldBe(32 * (256 * 1024 + 1));
        guard.IsQuarantined("hog", out _).ShouldBeFalse();
    }

    [Fact]
    public async Task A_script_that_throws_propagates_unwrapped_and_counts_as_a_completion()
    {
        var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 2 });
        const string throwing = "throw new System.InvalidOperationException(\"boom\");";
        using var thrower = Compile(throwing);
        // A spin that outlives the timeout until released, so the slot frees up between attempts.
        const string briefSpin = BriefSpinScript;
        using var spin = Compile(briefSpin);
        var ct = TestContext.Current.CancellationToken;
        var held = new FakeScriptContext();

        // timeout #1
        await Should.ThrowAsync<ScriptTimeoutException>(() => Run(guard, "k", briefSpin, spin, ShortTimeout, ct, held));
        await ReleaseAndDrain(guard, held);

        // A throw is a completion — the script returned control — so the consecutive counter resets…
        var ex = await Should.ThrowAsync<InvalidOperationException>(() => Run(guard, "k", throwing, thrower, ct: ct));
        ex.Message.ShouldBe("boom");
        guard.InFlightCount.ShouldBe(0);

        // …which is why this second timeout does NOT reach the threshold of 2.
        held.Released = false;
        await Should.ThrowAsync<ScriptTimeoutException>(() => Run(guard, "k", briefSpin, spin, ShortTimeout, ct, held));
        guard.IsQuarantined("k", out _).ShouldBeFalse();
        await ReleaseAndDrain(guard, held);
    }

    [Fact]
    public async Task Caller_cancellation_wins_over_the_timeout()
    {
        var guard = new ScriptRunGuard();
        using var spin = Compile(SpinScript);
        using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(80));

        await Should.ThrowAsync<OperationCanceledException>(() =>
            Run(guard, "k", SpinScript, spin, TimeSpan.FromSeconds(5), cts.Token));
        guard.IsQuarantined("k", out _).ShouldBeFalse();
    }

    [Fact]
    public async Task Quarantine_is_observable_through_the_event_and_the_counter()
    {
        var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 });
        using var spin = Compile(SpinScript);
        var seen = new List<(string Key, string Reason)>();
        guard.Quarantined += (_, e) => seen.Add((e.Key, e.Reason));

        await Should.ThrowAsync<ScriptQuarantinedException>(() =>
            Run(guard, "k", SpinScript, spin, ShortTimeout, TestContext.Current.CancellationToken));

        seen.ShouldHaveSingleItem();
        seen[0].Key.ShouldBe("k");
        seen[0].Reason.ShouldContain("consecutive");
        guard.QuarantinedCount.ShouldBe(1);
    }

    /// <summary>Releases the held script (signal), waits for the orphan to drain, and re-arms the gate for the next attempt.</summary>
    private static async Task ReleaseAndDrain(ScriptRunGuard guard, FakeScriptContext held)
    {
        held.Released = true;
        await WaitForIdle(guard);
        held.Released = false;
    }

    private static async Task WaitForIdle(ScriptRunGuard guard)
    {
        var deadline = DateTime.UtcNow.AddSeconds(30);
        while (guard.InFlightCount != 0 && DateTime.UtcNow < deadline)
            await Task.Delay(10);
        guard.InFlightCount.ShouldBe(0, "the brief spin must have finished");
    }
}