fix(scripting): quarantine a refusal behind a stuck orphan (fix round 2)

Ruling refinement on I1. Only timeouts are strikes, so a never-returning
while(true), which times out once and is then only refused, was never
quarantined, and that defeated E1. A refusal now quarantines immediately
when the orphan it was refused behind has been executing for longer than
ScriptingOptions.StuckOrphanQuarantineFactor x the script's timeout
(default 10, so 20 s at the 2 s default; 0 or less disables it). It uses
the same quarantine path: one Error, otopcua.script.quarantined with
reason=stuck, Bad, lifted on source change or redeploy. A young orphan
(a short overrun) is still a plain refusal.

The guard stamps the run's start (Stopwatch timestamp) when the pool
picks it up and clears it when the run really finishes.

Tests: the 30-trigger while(true) regression (paced) now ends
quarantined with reason "stuck", still with exactly one orphan and a
healthy sibling. Positive control: a 2x overrun is refused, not
quarantined, and evaluates again once the orphan completes. A factor of
0 disables the rule. Docs (ScriptEditor "Runtime safety", VirtualTags)
and the seam remarks are updated.

Refs #734

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT

src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs

@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using System.Diagnostics;
 using System.Diagnostics.Metrics;
 using Serilog;
 using ZB.MOM.WW.OtOpcUa.Commons.Observability;
@@ -27,7 +28,10 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Scripting;
 ///         consecutive timeouts</b> (only a timeout is a strike; a refusal is a Bad publish plus its counter
 ///         and leaves the streak untouched — on a 1 Hz trigger with a 2 s budget one 4 s overrun is a
 ///         timeout plus two refusals and must not quarantine; a run that returns — a value or a throw —
-///         resets the streak): no further dispatch, every attempt throws
+///         resets the streak), <b>or on a refusal behind a stuck orphan</b> — one that has been executing for
+///         longer than <see cref="ScriptingOptions.StuckOrphanQuarantineFactor"/> × the timeout (20 s at the
+///         2 s default), which is how a never-returning <c>while(true)</c> (one timeout, then only refusals)
+///         gets quarantined: no further dispatch, every attempt throws
 ///         <see cref="ScriptQuarantinedException"/>, one Error log, one <c>otopcua.script.quarantined</c>
 ///         increment, one <see cref="Quarantined"/> event. Lifted by a source change (the operator fixed the
 ///         script — a new source identity under the same key starts clean and the old orphan no longer blocks
@@ -63,11 +67,11 @@ public sealed class ScriptRunGuard
 
     /// <summary>
     ///     Counter <c>otopcua.script.quarantined</c> (Prometheus: <c>otopcua_script_quarantined</c>): one
-    ///     increment per script quarantined, tagged <c>reason</c> = <c>timeouts</c> | <c>allocation</c>.
+    ///     increment per script quarantined, tagged <c>reason</c> = <c>timeouts</c> | <c>stuck</c> | <c>allocation</c>.
     /// </summary>
     public static readonly Counter<long> QuarantinedCounter =
         OtOpcUaTelemetry.Meter.CreateCounter<long>("otopcua.script.quarantined", unit: "{script}",
-            description: "Scripts quarantined by the run guard (reason: timeouts | allocation).");
+            description: "Scripts quarantined by the run guard (reason: timeouts | stuck | allocation).");
 
     /// <summary>
     ///     Counter <c>otopcua.script.run_refused</c>: runs refused because the script's previous run was
@@ -80,6 +84,7 @@ public sealed class ScriptRunGuard
     private readonly ConcurrentDictionary<string, KeyState> _keys = new(StringComparer.Ordinal);
     private readonly long _maxAllocBytesPerRun;
     private readonly int _quarantineThreshold;
+    private readonly double _stuckFactor;
     private readonly ILogger _logger;
     private int _inFlight;
     private long _started;
@@ -94,6 +99,7 @@ public sealed class ScriptRunGuard
         options ??= new ScriptingOptions();
         _maxAllocBytesPerRun = options.MaxAllocBytesPerRun;
         _quarantineThreshold = Math.Max(1, options.QuarantineAfterConsecutiveTimeouts);
+        _stuckFactor = options.StuckOrphanQuarantineFactor;
         _logger = logger ?? Log.Logger;
     }
 
@@ -185,6 +191,24 @@ public sealed class ScriptRunGuard
             // ran long once. The refusal also leaves the timeout streak untouched.
             Interlocked.Increment(ref _refused);
             RefusedCounter.Add(1, new KeyValuePair<string, object?>("script", key));
+            // …unless the orphan is STUCK: a never-returning script times out once and is then only ever
+            // refused, so the timeout streak alone would never quarantine it. Past the stuck threshold the
+            // refusal itself quarantines (same path: Error log, metric, Bad, lifted on redeploy/source change).
+            if (_stuckFactor > 0 && Volatile.Read(ref state.RunStartedTimestamp) is var startedAt and not 0)
+            {
+                var running = Stopwatch.GetElapsedTime(startedAt);
+                var threshold = timeout * _stuckFactor;
+                if (running > threshold)
+                {
+                    Quarantine(key, state,
+                        $"its previous evaluation has been running for {running.TotalSeconds:F1} s, over " +
+                        $"{_stuckFactor:0.##}x its {timeout.TotalMilliseconds:F0} ms budget " +
+                        "(Scripting:StuckOrphanQuarantineFactor) — it is stuck; the script is not evaluated again " +
+                        "until its source changes or it is redeployed",
+                        "stuck");
+                    throw new ScriptQuarantinedException(key, state.QuarantineReason!);
+                }
+            }
             throw new ScriptRunRefusedException(key);
         }
 
@@ -200,6 +224,7 @@ public sealed class ScriptRunGuard
         // never complete, and the slot would never be released. The evaluator observes the token instead.
         var runTask = Task.Run(async () =>
         {
+            Volatile.Write(ref state.RunStartedTimestamp, Stopwatch.GetTimestamp());
             started.TrySetResult();
             var before = GC.GetAllocatedBytesForCurrentThread();
             try
@@ -223,6 +248,7 @@ public sealed class ScriptRunGuard
         // still quarantines its key here.
         _ = runTask.ContinueWith(t =>
         {
+            Volatile.Write(ref state.RunStartedTimestamp, 0);
             Interlocked.Exchange(ref state.InFlight, 0);
             Interlocked.Decrement(ref _inFlight);
             runCts.Dispose();
@@ -339,6 +365,8 @@ public sealed class ScriptRunGuard
         public string? SourceIdentity = sourceIdentity;
         public int InFlight;
         public int ConsecutiveTimeouts;
+        /// <summary><see cref="Stopwatch.GetTimestamp"/> when the pool started the in-flight run; 0 when idle.</summary>
+        public long RunStartedTimestamp;
         public string? QuarantineReason;
     }
 
@@ -373,7 +401,7 @@ public sealed class ScriptRunRefusedException : Exception
 
 /// <summary>
 ///     Thrown by <see cref="ScriptRunGuard.RunAsync{TContext, TResult}"/> for a quarantined script: repeated
-///     timeouts or an allocation-budget breach. Nothing is dispatched while quarantined; the seam
+///     timeouts, a stuck orphan or an allocation-budget breach. Nothing is dispatched while quarantined; the seam
 ///     publishes Bad carrying the reason, which an operator clears by fixing the script (source change) or
 ///     redeploying.
 /// </summary>