Merge #789 Windows worker pipe I/O off the completion port (+review fixes)

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

docs/plans/2026-11-09-out-of-process-script-runner-design.md

@@ -965,6 +965,7 @@ on Unix by the socket engine. Changes:
   - A full pipe blocks only the one thread draining the writer; every other producer just queues. A peer that stops
     reading is killed by the heartbeat check within 3 s, which fails that write.
   - Windows keeps overlapped async I/O, because a synchronous full-duplex handle serialises reads against writes.
+    (Superseded in §12.6: Windows now blocks too, through `WindowsBlockingPipeStream`.)
 - **Kill via `TerminateJobObject`** on Windows. The job admits one process, so no system-wide snapshot is needed.
 - Measured and **not** kept for memory, because none moved the compiled-worker RSS:
   - ReadyToRun worker publish: 195 MB;
@@ -1299,6 +1300,91 @@ Owner decisions §14 #5 and #6, as built. Operator reference: `docs/ServiceHosti
 - **Not verified live:** the installer on a real node. That is the QA deploy gate on lmxopcua#780's follow-up (see
   that issue for the checklist).
 
+### 12.6 lmxopcua#789 host CPU on Windows: blocking pipe I/O off the completion port (2026-11-11)
+
+**Already in place before this slice** (checked against the issue's candidates): one timer per evaluation for T and
+T + G (§12.2); the source SHA-256, its hex form and the negative-cache key are computed once per key's source
+(`KeyState.Sha`), not per `RunAsync`; batch submit (§12.3) is used by the Calculation pump, the VirtualTag coalescer and
+the alarm engine on the out-of-process path. The §8 paced harness phases call `RunAsync` one at a time on purpose (one
+trigger per evaluation); `Batch_throughput_and_alarm_pass_latency` measures the batched path.
+
+**What changed.**
+- **`WindowsBlockingPipeStream`** (Protocol). A .NET pipe opened `Asynchronous` completes even its synchronous
+  `Read`/`Write` through the thread pool's completion port, so every frame on Windows cost a completion packet plus a
+  pool work item. The stream issues its own overlapped `ReadFile`/`WriteFile` on the pipe handle with the
+  `OVERLAPPED.hEvent` low-order bit set, which tells the kernel not to queue a completion packet, and waits on that
+  event on the calling thread. Reads and writes stay concurrent (a handle opened for synchronous I/O would serialise
+  them). Dispose cancels a waiting read or write (`CancelIoEx`) and closes the pipe; a peer that goes away ends a read as
+  end of stream and fails a write with `IOException`, as `PipeStream` does.
+- **`ScriptWorkerFrameCodec.BlockingStream(pipe)`** picks that stream on Windows and the pipe itself on Unix. Both ends
+  now use the dedicated blocking reader thread and the synchronous combining writer on every OS. `UseBlockingIo` and the
+  two async read loops are gone; `HandshakeAsync` stays for its tests.
+- **1 MiB pipe buffer quota each way** (`WorkerPipeSecurity.BufferBytes`). With a quota of 0 an overlapped write stays
+  pending until the peer has read the bytes, so the blocking writer waited out a peer wake-up per frame. The first cut
+  without it halved the windev closed loop (54k → 26–27k evals/s); with it the closed loop is back.
+- Tests: `BlockingPipeStreamTests` (duplex with reads blocked on both ends, a 1 MB frame, peer gone, self-dispose wakes
+  the reader).
+
+**Measured on windev** (same VM as §12.1, idle, Release, `Throughput_and_latency_against_the_accepted_gate`, 10 s
+phases, base = `f109e5f2`, runs alternated base / new):
+
+| K = 2 | base (3 runs) | new (2 runs) | §8.2 budget |
+|---|---:|---:|---|
+| Host CPU at paced 10k/s, raw | 116 / 118 / 120 % | 61 / 84 % | — |
+| … vs InProcess (InProcess 6–9 %) | +108 / +112 / +112 | **+53 / +78** | ≤ 25 ✗ |
+| Host CPU at paced 5k/s, vs InProcess | +35 / +37 / +38 | −26 / −38 | — |
+| Idle p50 (sequential) | 491 / 499 / 560 µs | 363 / 386 µs | ≤ 150 µs ✗ |
+| Idle host CPU (sequential) | 57 / 61 / 66 % | 27 / 37 % | — |
+| p99 at 10k/s | 2,108 / 2,264 / 2,373 µs | 878 / 944 µs | ≤ 2 ms ✓ (was ✗) |
+| Closed loop ×64 | 51k / 54k / 55k evals/s | 54k / 55k evals/s | gate ≥ 5k ✓ |
+
+- The first cut (no buffer quota), two runs: 10k/s raw 118 / 127 %, idle p50 354 / 362 µs, closed loop 26–27k/s.
+- **Attribution of what is left.** With `DOTNET_ThreadPool_UnfairSemaphoreSpinLimit=0` the new build reads 71 % raw at
+  10k/s (+65) and the base 93 % (+85). So pool spinning is no longer what the remainder is made of: it is about 60 µs of
+  host CPU per evaluation in syscalls and thread wake-ups on this VM (reader wake, dispatcher wake, and the harness's
+  own `ContinueWith` pool hop per result), against about 13 µs on macOS. A non-VM Windows box is still unmeasured.
+
+**macOS** (Unix path unchanged; 2 runs each, alternated, load average 3–14): K = 2 at 10k/s +9 / +12 vs base +16 / +5
+(InProcess 46–57 %); idle p50 26 µs both; closed loop 365–374k vs 364–379k evals/s. No change, as expected.
+
+**Tests.** ScriptWorker `Category=ScriptWorker` on windev: 107 passed, 2 skipped by design (109), including all
+`WorkerSoftClockTests` suspend cases, the deferred-start / runaway-in-batch deadline tests, the kill and quarantine tests
+and the restricted-token worker. On macOS: 104 passed, 5 skipped (109). Unit classes `BlockingPipeStreamTests`,
+`ProtocolTests` and `HandshakeTests`: 67 / 67 on both.
+
+**Review follow-ups (same slice).** Blocking I/O moves two failure modes onto dedicated threads and producers:
+- **Reader faults.** The host reader runs completion work inline (slot release, load bookkeeping). The frame read and
+  its handling are now caught separately: only the read's I/O exceptions mean the pipe ended. Anything the handling
+  throws kills that worker as `ProtocolFault`, so its outstanding runs answer "worker restarted" (BadResourceUnavailable)
+  and the shard restarts. Before, an exception escaping a dedicated thread would end the host process, and an
+  `InvalidOperationException` was swallowed as a clean pipe end, leaving the worker unread until a `Crash` 2 s later.
+  The worker's reader has the same catch-all (exit `ProtocolFault`).
+- **A worker that talks but does not listen.** A full pipe blocks the producer that is draining the writer: a seam
+  submitting, a deploy loading scripts, a Cancel. Heartbeats keep the silence check quiet, and with only loads queued
+  nothing else fires. The heartbeat monitor now kills a worker whose pipe write has been blocked for longer than
+  `max(10 s, silence limit)` (`ScriptWorkerFrameWriter.WriteInProgressFor`, `KilledUnresponsive`). That fails the write
+  and releases the producer. The one known way a healthy worker's reader could wedge is closed too: `Cancel` now runs a
+  script's `ctx.CancellationToken` callbacks on the pool, not on the worker's reader. `StopAsync` sends `Shutdown` off
+  the caller, so its 1 s budget holds on a full pipe.
+- `WindowsBlockingPipeStream` refuses the async surface, because `Stream`'s base versions serialise a write behind a
+  waiting read.
+- Tests: `PipeFaultTests` covers a throwing completion callback (the worker ends `ProtocolFault`, the sibling in flight
+  gets BadResourceUnavailable, and the shard serves again) and a worker that stops reading while it heartbeats (killed
+  `KilledUnresponsive` within the bound, and the blocked deploy producer returns). The second uses the
+  `Hello.test_stall_reader_on_diag` seam. `BlockingPipeStreamTests` adds dispose with a write blocked on a full pipe,
+  and with a read and a write both pending. The protocol, handshake and pipe classes now carry `Category=ScriptWorker`.
+  Mutation checks: with no stall kill the stall test is red (timeout); with a catch-all that does not fault, the
+  callback test is red (`Crash`).
+- Re-run: `Category=ScriptWorker` on windev: 179 passed, 2 skipped by design (181). On macOS: 176 passed, 5 skipped
+  (181). windev perf after the fixes (alternated new / base / new): 10k/s vs InProcess +55 / +68 (base +95 in the same
+  session); idle p50 391 / 365 µs (base 517); closed loop 51k / 54k evals/s (base 57k). No regression against the
+  figures above.
+
+**Still missed:** Windows host CPU vs InProcess (+53 to +78 against ≤ 25) and Windows idle p50 (363–386 µs against
+≤ 150 µs; InProcess alone is about 105 µs on this VM). Next candidates: drop the dispatcher wake when the caller's
+continuation is known to be cheap (needs a seam-level guarantee, the reason it exists is a slow consumer), or measure
+on non-VM Windows hardware before tuning further.
+
 ---
 
 ## 13. Test plan

src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs

@@ -91,6 +91,10 @@ internal sealed record BatchSubmission(string Key, EvalRequest Request, TimeSpan
 ///     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="TestStallReaderOnDiag">
+///     Test seam: the worker's reader stops reading for good at the first diagnostics request while its heartbeats go on —
+///     a worker that keeps talking but stopped listening (the write-stall check, lmxopcua#789).
+/// </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.
@@ -110,7 +114,8 @@ internal sealed record WorkerLaunchSpec(
     int CompileThreads = 0,
     string? WorkerSid = null,
     bool WorkerSoftClock = true,
-    TimeSpan? HeartbeatSilenceLimit = null)
+    TimeSpan? HeartbeatSilenceLimit = null,
+    bool TestStallReaderOnDiag = false)
 {
     /// <summary>The default compile timeout.</summary>
     public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60);
@@ -138,7 +143,11 @@ internal sealed class WorkerConnection : IAsyncDisposable
     ];
 
     private readonly Process _process;
-    private readonly NamedPipeServerStream _pipe;
+    /// <summary>
+    ///     The pipe as both directions drive it: blocking, through <see cref="ScriptWorkerFrameCodec.BlockingStream"/>
+    ///     (the pipe itself on Unix, a <see cref="WindowsBlockingPipeStream"/> over it on Windows). Owns the pipe.
+    /// </summary>
+    private readonly Stream _pipe;
     private readonly IDisposable? _job;
     private readonly IDisposable? _stdio;
     private readonly string _workDir;
@@ -171,7 +180,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
     private long _nextLoadId;
     private int _disposed;
 
-    private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, IDisposable? stdio,
+    private WorkerConnection(string shard, Process process, Stream 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)
     {
@@ -192,7 +201,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
         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);
+            synchronous: true);
         _dispatcher = new ResultDispatcher($"ScriptWorker[{shard}].Results");
     }
 
@@ -261,6 +270,20 @@ internal sealed class WorkerConnection : IAsyncDisposable
     /// </summary>
     internal TimeSpan? ProgressCeilingOverride { get; set; }
 
+    /// <summary>
+    ///     How long one pipe write may stay blocked before the worker is killed (<see cref="WorkerExitReason.KilledUnresponsive"/>,
+    ///     lmxopcua#789). Writes are blocking, so a full pipe — a worker whose reader stopped while its heartbeats go on —
+    ///     would otherwise hold whichever producer is draining the writer (a seam submitting, a deploy loading scripts, a
+    ///     Cancel) forever: the silence check never fires, and with only loads queued nothing else does. A worker that is
+    ///     reading drains the 1 MiB buffer far faster than this. Null = the larger of 10 s and the silence limit; set = an
+    ///     override (tests).
+    /// </summary>
+    internal TimeSpan? WriteStallLimitOverride { get; set; }
+
+    /// <summary>The write-stall bound now (see <see cref="WriteStallLimitOverride"/>).</summary>
+    internal TimeSpan WriteStallLimit
+        => WriteStallLimitOverride ?? (SilenceLimit > TimeSpan.FromSeconds(10) ? SilenceLimit : TimeSpan.FromSeconds(10));
+
     /// <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; }
 
@@ -312,6 +335,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
             ? WorkerPipeSecurity.CreateServer(pipeName, WorkerPipeSecurity.ForCurrentHost(spec.WorkerSid))
             : new NamedPipeServerStream(pipeName, PipeDirection.InOut, maxNumberOfServerInstances: 1,
                 PipeTransmissionMode.Byte, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
+        Stream? io = null;
         Process? process = null;
         IDisposable? job = null;
         IDisposable? stdio = null;
@@ -375,17 +399,17 @@ internal sealed class WorkerConnection : IAsyncDisposable
                 MaxLogLines = 1000,
                 CompileThreads = (uint)Math.Max(0, spec.CompileThreads),
                 WorkerSoftClock = spec.WorkerSoftClock,
+                TestStallReaderOnDiag = spec.TestStallReaderOnDiag,
             };
             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);
+            io = ScriptWorkerFrameCodec.BlockingStream(pipe);
+            var ready = await HandshakeBlockingAsync(io, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false);
 
-            var connection = new WorkerConnection(spec.Shard, process, pipe, job, stdio, workDir, log, onLogs, spec.HeartbeatInterval,
+            var connection = new WorkerConnection(spec.Shard, process, io, job, stdio, workDir, log, onLogs, spec.HeartbeatInterval,
                 spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock,
                 heartbeatSilenceLimit: spec.HeartbeatSilenceLimit);
             connection.Begin();
@@ -397,7 +421,8 @@ internal sealed class WorkerConnection : IAsyncDisposable
             job?.Dispose();
             stdio?.Dispose();
             process?.Dispose();
-            await pipe.DisposeAsync().ConfigureAwait(false);
+            if (io is not null) await io.DisposeAsync().ConfigureAwait(false);
+            else await pipe.DisposeAsync().ConfigureAwait(false);
             TryDeleteDirectory(workDir);
             WorkerLaunchMarker.TryDelete(markerPath);
             throw;
@@ -430,7 +455,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
     }
 
     /// <summary>
-    ///     As <see cref="HandshakeAsync"/> with blocking I/O only (<see cref="ScriptWorkerFrameCodec.UseBlockingIo"/>):
+    ///     As <see cref="HandshakeAsync"/> with blocking I/O only (<see cref="ScriptWorkerFrameCodec.BlockingStream"/>):
     ///     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.
@@ -548,8 +573,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
         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();
+        new Thread(ReadLoopBlocking) { IsBackground = true, Name = $"ScriptWorker[{Shard}].Reader" }.Start();
         _ = Task.Run(HeartbeatMonitorAsync);
         _ = Task.Run(WatchProcessAsync);
     }
@@ -751,7 +775,9 @@ internal sealed class WorkerConnection : IAsyncDisposable
         if (!_exited.Task.IsCompleted)
         {
             lock (_pending) { _killedFor ??= (reason, null); }
-            Send(new Envelope { Shutdown = new Shutdown() });
+            // Off this thread: the send may drain the writer, and a write blocks on a full pipe — the 1 s budget below
+            // must hold regardless (the kill then fails that write).
+            _ = Task.Run(() => Send(new Envelope { Shutdown = new Shutdown() }));
             try
             {
                 await _process.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
@@ -785,50 +811,56 @@ internal sealed class WorkerConnection : IAsyncDisposable
     }
 
     /// <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.
+    ///     The reader, on a dedicated thread with blocking reads (<see cref="ScriptWorkerFrameCodec.BlockingStream"/>,
+    ///     lmxopcua#789): a frame wakes this thread straight from the kernel, instead of the socket engine's epoll thread
+    ///     (Unix) or the completion port (Windows) queueing a thread-pool work item per read.
     /// </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)
+        while (true)
         {
+            Envelope? frame;
+            try
+            {
+                frame = reader.Read();
+            }
+            catch (ScriptWorkerProtocolException ex)
+            {
+                fault = ProtocolFault(ex);
+                break;
+            }
+            catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException
+                                           or System.Net.Sockets.SocketException)
+            {
+                break; // the pipe ended (these come from the read only, never from the frame's handling below)
+            }
+            if (frame is null) break;
+
+            try
+            {
+                HandleFrame(frame);
+            }
+            catch (ScriptWorkerProtocolException ex)
+            {
+                fault = ProtocolFault(ex);
+                break;
+            }
+            catch (Exception ex)
+            {
+                // HandleFrame runs completion work inline (slot releases, load bookkeeping). On this dedicated thread an
+                // exception that escaped would take the whole server process down, and treating it as the pipe ending
+                // would leave the worker running unread; so it ends this worker as a fault: every outstanding evaluation
+                // completes as "worker restarted" (BadResourceUnavailable) and the shard restarts.
+                _log.Error(ex, "ScriptWorker[{Shard}] failed handling a {Frame} frame; killing pid {Pid}", Shard, frame.BodyCase, Pid);
+                fault = WorkerExitReason.ProtocolFault;
+                break;
+            }
         }
         _ = 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);
@@ -958,6 +990,15 @@ internal sealed class WorkerConnection : IAsyncDisposable
                     Kill(WorkerExitReason.KilledHeartbeat, null);
                     return;
                 }
+                if (_writer.WriteInProgressFor is { } blocked && blocked > WriteStallLimit)
+                {
+                    // It talks but does not listen: the pipe is full and a producer is blocked in the write. The kill
+                    // fails that write and releases it (lmxopcua#789).
+                    _log.Warning("ScriptWorker[{Shard}] pid {Pid} has not read from its pipe for {Blocked} ms (a write is blocked on a full pipe) while still heartbeating; killing it",
+                        Shard, Pid, (long)blocked.TotalMilliseconds);
+                    Kill(WorkerExitReason.KilledUnresponsive, null);
+                    return;
+                }
                 if (!_pending.IsEmpty && ProgressCeiling() is { } ceiling)
                 {
                     var stuck = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastProgressTimestamp));

src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerPipeSecurity.cs

@@ -50,11 +50,21 @@ internal static class WorkerPipeSecurity
         return security;
     }
 
+    /// <summary>
+    ///     The pipe's buffer quota in each direction (lmxopcua#789). With a quota of 0 an overlapped write stays pending
+    ///     until the peer has READ the bytes, so a blocking writer (<see cref="ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.WindowsBlockingPipeStream"/>)
+    ///     would wait out a full peer wake-up per frame — measured on windev as half the closed-loop throughput. With a
+    ///     quota a write completes as soon as the bytes are buffered; a peer that stops reading still fills it and blocks
+    ///     the writer: the host's write-stall check (<c>WorkerConnection.WriteStallLimit</c>) then kills the worker, which
+    ///     fails the write and releases the blocked thread.
+    /// </summary>
+    internal const int BufferBytes = 1024 * 1024;
+
     /// <summary>Creates the single-instance server end of a worker pipe with <paramref name="security"/>.</summary>
     /// <param name="pipeName">The random pipe name.</param>
     /// <param name="security">The ACL (<see cref="ForCurrentHost"/>).</param>
     /// <returns>The server stream.</returns>
     public static NamedPipeServerStream CreateServer(string pipeName, PipeSecurity security)
         => NamedPipeServerStreamAcl.Create(pipeName, PipeDirection.InOut, maxNumberOfServerInstances: 1,
-            PipeTransmissionMode.Byte, PipeOptions.Asynchronous, inBufferSize: 0, outBufferSize: 0, security);
+            PipeTransmissionMode.Byte, PipeOptions.Asynchronous, inBufferSize: BufferBytes, outBufferSize: BufferBytes, security);
 }

src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto

@@ -45,6 +45,9 @@ message Hello {
   // starts the run, and to report the crossing in EvalResults.soft_expired_eval_ids. The host then keeps only a
   // transport deadline of its own (see EvalResults). Honoured only when the worker echoes Ready.worker_soft_clock.
   bool worker_soft_clock = 11;
+  // Test seam (lmxopcua#789): the worker's reader stops reading for good at the first DiagnosticsRequest while its
+  // heartbeats go on, so the host's write-stall check can be exercised against a real worker. Never set in production.
+  bool test_stall_reader_on_diag = 12;
 }
 
 message Ready {

src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFrameCodec.cs

@@ -78,14 +78,21 @@ public static class ScriptWorkerFrameCodec
     }
 
     /// <summary>
-    ///     Whether this platform's pipe should be driven with blocking I/O (lmxopcua#789). On Unix a .NET pipe is a
-    ///     socket, and ONE asynchronous operation on it switches it to non-blocking for good, after which even a
-    ///     "blocking" read goes through the socket engine (an epoll-thread hand-off and a spinning wait per frame) —
-    ///     measured as most of the host's per-evaluation CPU. Kept blocking, a read is a plain kernel wait on a
-    ///     dedicated thread. Not on Windows, where an overlapped pipe handle is needed for full duplex and its
-    ///     asynchronous path is the efficient one.
+    ///     The stream both ends drive a connected worker pipe through, with blocking I/O on dedicated threads
+    ///     (lmxopcua#789). On Unix a .NET pipe is a socket, and ONE asynchronous operation on it switches it to
+    ///     non-blocking for good, after which even a "blocking" read goes through the socket engine (an epoll-thread
+    ///     hand-off and a spinning wait per frame) — measured as most of the host's per-evaluation CPU; so the pipe itself
+    ///     is used, only ever synchronously. On Windows an overlapped pipe completes even its synchronous calls through the
+    ///     thread pool's completion port, so the pipe is wrapped in a <see cref="WindowsBlockingPipeStream"/>, which waits
+    ///     for its own overlapped I/O without the port and keeps reads and writes concurrent.
     /// </summary>
-    public static bool UseBlockingIo => !OperatingSystem.IsWindows();
+    /// <param name="pipe">The connected pipe (opened with <see cref="System.IO.Pipes.PipeOptions.Asynchronous"/>).</param>
+    /// <returns>The stream to read and write frames through; disposing it closes the pipe.</returns>
+    public static Stream BlockingStream(System.IO.Pipes.PipeStream pipe)
+    {
+        ArgumentNullException.ThrowIfNull(pipe);
+        return OperatingSystem.IsWindows() ? new WindowsBlockingPipeStream(pipe) : pipe;
+    }
 
     /// <summary>Writes one frame synchronously (the blocking-I/O handshake). Not thread-safe.</summary>
     /// <param name="stream">The pipe.</param>

src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs

@@ -122,10 +122,12 @@ public sealed class ScriptWorkerFrameReader
 /// </summary>
 /// <remarks>
 ///     <para>Asynchronous mode: a write that cannot complete at once (a full pipe) continues on the I/O completion, and
-///     frames queued meanwhile wait for it. Synchronous mode (<see cref="ScriptWorkerFrameCodec.UseBlockingIo"/>, so
-///     the socket under a Unix pipe never turns non-blocking): the draining thread blocks while the pipe is full. Only
+///     frames queued meanwhile wait for it. Synchronous mode (the worker pipe, through
+///     <see cref="ScriptWorkerFrameCodec.BlockingStream"/>, so no frame needs the socket engine or the completion port):
+///     the draining thread blocks while the pipe is full. Only
 ///     one thread drains at a time — every other producer just queues and returns — and a peer that stops reading is
-///     killed by its supervisor (the host's heartbeat check), which fails the write and releases that thread.</para>
+///     killed by its supervisor (the host's write-stall check, <see cref="WriteInProgressFor"/>; on the worker side the
+///     host's silence check), which fails the write and releases that thread.</para>
 ///     <para>A failed write completes the writer and reports the exception once through <c>onFault</c>.</para>
 /// </remarks>
 public sealed class ScriptWorkerFrameWriter
@@ -140,6 +142,9 @@ public sealed class ScriptWorkerFrameWriter
     private int _faulted;
     private long _seq;
 
+    /// <summary><see cref="System.Diagnostics.Stopwatch.GetTimestamp"/> when the write in progress started; 0 = none.</summary>
+    private long _writeStarted;
+
     /// <summary>Initializes the writer.</summary>
     /// <param name="stream">The pipe.</param>
     /// <param name="onFault">Called once when a write fails (or a frame is refused); the writer is then completed.</param>
@@ -161,6 +166,14 @@ public sealed class ScriptWorkerFrameWriter
     /// <summary>The negotiated frame cap.</summary>
     public int MaxFrameBytes { get; set; }
 
+    /// <summary>
+    ///     How long the pipe write in progress has been blocked, or null when no write is in progress. A blocking write
+    ///     stays in progress while the pipe is full, i.e. while the peer is not reading: the supervisor uses this to end a
+    ///     peer that keeps talking but stopped listening (lmxopcua#789), which releases the thread blocked here.
+    /// </summary>
+    public TimeSpan? WriteInProgressFor
+        => Volatile.Read(ref _writeStarted) is var started and not 0 ? System.Diagnostics.Stopwatch.GetElapsedTime(started) : null;
+
     /// <summary>Whether <see cref="Complete"/> has been called or a write failed.</summary>
     public bool IsCompleted => Volatile.Read(ref _completed) != 0;
 
@@ -195,8 +208,16 @@ public sealed class ScriptWorkerFrameWriter
                 FillPass();
                 if (_pass.WrittenCount > 0)
                 {
-                    if (_synchronous) _stream.Write(_pass.WrittenSpan);
-                    else await _stream.WriteAsync(_pass.WrittenMemory).ConfigureAwait(false);
+                    Volatile.Write(ref _writeStarted, System.Diagnostics.Stopwatch.GetTimestamp());
+                    try
+                    {
+                        if (_synchronous) _stream.Write(_pass.WrittenSpan);
+                        else await _stream.WriteAsync(_pass.WrittenMemory).ConfigureAwait(false);
+                    }
+                    finally
+                    {
+                        Volatile.Write(ref _writeStarted, 0);
+                    }
                     _pass.ResetWrittenCount();
                     // Synchronous on purpose: a pipe has no write buffer to flush, and PipeStream does not override
                     // FlushAsync, so the base Stream.FlushAsync would queue a thread-pool work item per pass.

src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/WindowsBlockingPipeStream.cs

@@ -0,0 +1,307 @@
+using System.ComponentModel;
+using System.IO.Pipes;
+using System.Runtime.InteropServices;
+using System.Runtime.Versioning;
+using Microsoft.Win32.SafeHandles;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
+
+/// <summary>
+///     Blocking, full-duplex I/O on an overlapped Windows pipe handle that never touches the I/O completion port
+///     (lmxopcua#789).
+/// </summary>
+/// <remarks>
+///     <para>A .NET pipe opened with <see cref="PipeOptions.Asynchronous"/> is bound to the thread pool's completion
+///     port, and even its synchronous <c>Read</c>/<c>Write</c> complete through it: every frame costs a completion-port
+///     packet plus a thread-pool work item, and the pool's spin-waiting between those items was most of the host's CPU
+///     per evaluation on Windows (+115 points over in process at 10k evals/s). A handle opened for synchronous I/O
+///     would avoid the port but serialises a read against a write on the same file object, which a full-duplex pipe
+///     with a blocked reader cannot afford.</para>
+///     <para>So this stream issues its own overlapped <c>ReadFile</c>/<c>WriteFile</c> on the pipe's handle with an
+///     <c>OVERLAPPED.hEvent</c> whose low-order bit is set — which tells the kernel not to queue a completion packet
+///     (documented on <c>GetQueuedCompletionStatus</c>) — and waits on that event on the calling thread. Reads and
+///     writes stay concurrent (one read and one write at a time, which is how the frame reader and the combining writer
+///     use a pipe), and a frame wakes the waiting thread straight from the kernel: the Windows twin of the Unix
+///     blocking socket (<see cref="ScriptWorkerFrameCodec.BlockingStream"/>).</para>
+///     <para>Disposing cancels any read or write still waiting (<c>CancelIoEx</c>) and then closes the pipe; the waiter
+///     sees end of stream (a read) or an <see cref="ObjectDisposedException"/>. A peer that goes away completes a
+///     pending read as end of stream and fails a pending write with an <see cref="IOException"/>, as
+///     <see cref="PipeStream"/> does.</para>
+/// </remarks>
+[SupportedOSPlatform("windows")]
+public sealed unsafe class WindowsBlockingPipeStream : Stream
+{
+    private const int ErrorHandleEof = 38;
+    private const int ErrorBrokenPipe = 109;
+    private const int ErrorNoData = 232;
+    private const int ErrorPipeNotConnected = 233;
+    private const int ErrorOperationAborted = 995;
+    private const int ErrorIoPending = 997;
+
+    private const uint StatusSuccess = 0;
+    private const uint StatusEndOfFile = 0xC0000011;
+    private const uint StatusPipeDisconnected = 0xC00000B0;
+    private const uint StatusPipeClosing = 0xC00000B1;
+    private const uint StatusPipeBroken = 0xC000014B;
+    private const uint StatusCancelled = 0xC0000120;
+
+    private readonly PipeStream _pipe;
+    private readonly SafePipeHandle _handle;
+    private readonly Operation _read;
+    private readonly Operation _write;
+
+    /// <summary>The owner's reference plus one per operation in progress; the native state is freed when it reaches 0.</summary>
+    private int _refs = 1;
+    private int _disposed;
+
+    /// <summary>Wraps a connected pipe; the wrapper owns it from here on (disposing the wrapper closes it).</summary>
+    /// <param name="pipe">A connected pipe opened with <see cref="PipeOptions.Asynchronous"/>.</param>
+    public WindowsBlockingPipeStream(PipeStream pipe)
+    {
+        ArgumentNullException.ThrowIfNull(pipe);
+        if (!pipe.IsAsync) throw new ArgumentException("the pipe must be opened for overlapped (asynchronous) I/O", nameof(pipe));
+        _pipe = pipe;
+        _handle = pipe.SafePipeHandle;
+        _read = new Operation();
+        _write = new Operation();
+    }
+
+    /// <inheritdoc />
+    public override bool CanRead => Volatile.Read(ref _disposed) == 0;
+
+    /// <inheritdoc />
+    public override bool CanWrite => Volatile.Read(ref _disposed) == 0;
+
+    /// <inheritdoc />
+    public override bool CanSeek => false;
+
+    /// <inheritdoc />
+    public override long Length => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    public override long Position
+    {
+        get => throw new NotSupportedException();
+        set => throw new NotSupportedException();
+    }
+
+    /// <inheritdoc />
+    public override int Read(byte[] buffer, int offset, int count)
+    {
+        ValidateBufferArguments(buffer, offset, count);
+        return Read(buffer.AsSpan(offset, count));
+    }
+
+    /// <inheritdoc />
+    /// <returns>The bytes read; 0 at end of stream (the peer closed or went away).</returns>
+    public override int Read(Span<byte> buffer)
+    {
+        if (buffer.IsEmpty) return 0;
+        Enter();
+        try
+        {
+            fixed (byte* p = buffer)
+            {
+                var ov = _read.Prepare();
+                var started = ReadFile(_handle, p, buffer.Length, IntPtr.Zero, ov);
+                var (status, bytes) = _read.Finish(started, out var error);
+                if (error != 0)
+                {
+                    return error is ErrorBrokenPipe or ErrorPipeNotConnected or ErrorNoData or ErrorHandleEof
+                        ? 0
+                        : throw Failure(error, "read");
+                }
+                return status switch
+                {
+                    StatusSuccess => bytes,
+                    StatusEndOfFile or StatusPipeBroken or StatusPipeDisconnected or StatusPipeClosing => 0,
+                    _ => throw StatusFailure(status, "read"),
+                };
+            }
+        }
+        finally
+        {
+            Exit();
+        }
+    }
+
+    /// <inheritdoc />
+    public override void Write(byte[] buffer, int offset, int count)
+    {
+        ValidateBufferArguments(buffer, offset, count);
+        Write(new ReadOnlySpan<byte>(buffer, offset, count));
+    }
+
+    /// <inheritdoc />
+    public override void Write(ReadOnlySpan<byte> buffer)
+    {
+        Enter();
+        try
+        {
+            fixed (byte* start = buffer)
+            {
+                var done = 0;
+                while (done < buffer.Length)
+                {
+                    var ov = _write.Prepare();
+                    var started = WriteFile(_handle, start + done, buffer.Length - done, IntPtr.Zero, ov);
+                    var (status, bytes) = _write.Finish(started, out var error);
+                    if (error != 0) throw Failure(error, "write");
+                    if (status != StatusSuccess) throw StatusFailure(status, "write");
+                    if (bytes <= 0) throw new IOException("pipe write made no progress");
+                    done += bytes;
+                }
+            }
+        }
+        finally
+        {
+            Exit();
+        }
+    }
+
+    /// <summary>A no-op: a pipe has no write buffer to flush.</summary>
+    public override void Flush() { }
+
+    /// <summary>A no-op, completed: a pipe has no write buffer to flush.</summary>
+    /// <param name="cancellationToken">Unused.</param>
+    /// <returns>A completed task.</returns>
+    public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+
+    // The asynchronous surface is refused: Stream's base versions run the blocking call on the pool behind ONE semaphore
+    // shared by reads and writes, which would serialise a write behind a read that is waiting for the peer — the very
+    // deadlock this stream exists to avoid. Callers use the blocking calls on their own threads.
+
+    /// <inheritdoc />
+    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => throw AsyncRefused();
+
+    /// <inheritdoc />
+    public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) => throw AsyncRefused();
+
+    /// <inheritdoc />
+    public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => throw AsyncRefused();
+
+    /// <inheritdoc />
+    public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default) => throw AsyncRefused();
+
+    /// <inheritdoc />
+    public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) => throw AsyncRefused();
+
+    /// <inheritdoc />
+    public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) => throw AsyncRefused();
+
+    private static NotSupportedException AsyncRefused()
+        => new("WindowsBlockingPipeStream is blocking only: read and write on dedicated threads (lmxopcua#789)");
+
+    /// <inheritdoc />
+    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    public override void SetLength(long value) => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    protected override void Dispose(bool disposing)
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) == 0)
+        {
+            // Wake a reader or writer still waiting on the kernel before the handle goes: it then sees cancellation
+            // (reported as disposed), and the pipe's own close cannot race an operation that is still being issued.
+            try
+            {
+                var add = false;
+                _handle.DangerousAddRef(ref add);
+                try { _ = CancelIoEx(_handle.DangerousGetHandle(), IntPtr.Zero); }
+                finally { if (add) _handle.DangerousRelease(); }
+            }
+            catch (ObjectDisposedException) { }
+            _pipe.Dispose();
+            Exit();
+        }
+        base.Dispose(disposing);
+    }
+
+    private void Enter()
+    {
+        while (true)
+        {
+            var refs = Volatile.Read(ref _refs);
+            if (refs == 0 || Volatile.Read(ref _disposed) != 0) throw new ObjectDisposedException(nameof(WindowsBlockingPipeStream));
+            if (Interlocked.CompareExchange(ref _refs, refs + 1, refs) == refs) return;
+        }
+    }
+
+    private void Exit()
+    {
+        if (Interlocked.Decrement(ref _refs) != 0) return;
+        _read.Free();
+        _write.Free();
+    }
+
+    private Exception Failure(int error, string what)
+        => error == ErrorOperationAborted && Volatile.Read(ref _disposed) != 0
+            ? new ObjectDisposedException(nameof(WindowsBlockingPipeStream))
+            : new IOException($"pipe {what} failed: {new Win32Exception(error).Message}", new Win32Exception(error));
+
+    private Exception StatusFailure(uint status, string what)
+        => status == StatusCancelled && Volatile.Read(ref _disposed) != 0
+            ? new ObjectDisposedException(nameof(WindowsBlockingPipeStream))
+            : new IOException($"pipe {what} failed: NTSTATUS 0x{status:X8}");
+
+    /// <summary>One direction's native state: its OVERLAPPED and the manual-reset event it completes on.</summary>
+    private sealed class Operation
+    {
+        private readonly ManualResetEvent _done = new(false);
+        private readonly NativeOverlapped* _ov = (NativeOverlapped*)NativeMemory.AllocZeroed((nuint)sizeof(NativeOverlapped));
+        private readonly IntPtr _event;
+
+        public Operation()
+        {
+            // Low-order bit set: the completion is NOT queued to the completion port the pipe is bound to.
+            _event = _done.SafeWaitHandle.DangerousGetHandle() | 1;
+        }
+
+        public NativeOverlapped* Prepare()
+        {
+            *_ov = default;
+            _ov->EventHandle = _event;
+            return _ov;
+        }
+
+        /// <summary>Waits for the operation <paramref name="started"/> issued and returns its NTSTATUS and byte count.</summary>
+        /// <param name="started">What ReadFile / WriteFile returned.</param>
+        /// <param name="error">The Win32 error when the call failed outright (not pending); 0 otherwise.</param>
+        public (uint Status, int Bytes) Finish(bool started, out int error)
+        {
+            error = 0;
+            if (!started)
+            {
+                var last = Marshal.GetLastPInvokeError();
+                if (last != ErrorIoPending)
+                {
+                    error = last;
+                    return (0, 0);
+                }
+                // The kernel sets the event when the I/O completes (Read/WriteFile reset it when they start it).
+                _done.WaitOne();
+            }
+            return ((uint)(long)_ov->InternalLow, (int)(long)_ov->InternalHigh);
+        }
+
+        public void Free()
+        {
+            NativeMemory.Free(_ov);
+            _done.Dispose();
+        }
+    }
+
+    [DllImport("kernel32.dll", SetLastError = true)]
+    [return: MarshalAs(UnmanagedType.Bool)]
+    private static extern bool ReadFile(SafePipeHandle handle, byte* buffer, int count, IntPtr bytesRead, NativeOverlapped* overlapped);
+
+    [DllImport("kernel32.dll", SetLastError = true)]
+    [return: MarshalAs(UnmanagedType.Bool)]
+    private static extern bool WriteFile(SafePipeHandle handle, byte* buffer, int count, IntPtr bytesWritten, NativeOverlapped* overlapped);
+
+    [DllImport("kernel32.dll", SetLastError = true)]
+    [return: MarshalAs(UnmanagedType.Bool)]
+    private static extern bool CancelIoEx(IntPtr handle, IntPtr overlapped);
+}

src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.csproj

@@ -5,6 +5,8 @@
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
         <!-- protoc-generated members carry no XML docs. -->
         <NoWarn>$(NoWarn);CS1591</NoWarn>
+        <!-- WindowsBlockingPipeStream issues its own overlapped pipe I/O (lmxopcua#789). -->
+        <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     </PropertyGroup>
 
     <!-- lmxopcua#780: the host <-> ScriptWorker frame protocol (design §4). Shared by the host-side runner and the

src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/Program.cs

@@ -22,9 +22,8 @@ internal static class Program
             PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
         try
         {
-            // Synchronous on Unix so the socket under the pipe stays blocking (ScriptWorkerFrameCodec.UseBlockingIo).
-            if (ScriptWorkerFrameCodec.UseBlockingIo) pipe.Connect(TimeSpan.FromSeconds(10));
-            else await pipe.ConnectAsync(TimeSpan.FromSeconds(10), CancellationToken.None);
+            // Synchronous, so the socket under a Unix pipe stays blocking (ScriptWorkerFrameCodec.BlockingStream).
+            pipe.Connect(TimeSpan.FromSeconds(10));
         }
         catch (Exception ex)
         {
@@ -32,7 +31,7 @@ internal static class Program
             return ScriptWorkerEnvironment.ExitProtocolFault;
         }
 
-        var host = new WorkerHost(pipe, nonce, parentPid);
+        var host = new WorkerHost(ScriptWorkerFrameCodec.BlockingStream(pipe), nonce, parentPid);
         return await host.RunAsync();
     }
 }

src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs

@@ -32,7 +32,8 @@ internal sealed class WorkerHost
     /// <summary>The watchdog's longest sleep between passes.</summary>
     private static readonly TimeSpan WatchdogPoll = TimeSpan.FromMilliseconds(50);
 
-    private readonly NamedPipeClientStream _pipe;
+    /// <summary>The host pipe, driven blocking (<see cref="ScriptWorkerFrameCodec.BlockingStream"/>).</summary>
+    private readonly Stream _pipe;
     private readonly string _nonce;
     private readonly int _parentPid;
     private readonly ConcurrentDictionary<string, LoadedScript> _scripts = new(StringComparer.Ordinal);
@@ -93,7 +94,10 @@ internal sealed class WorkerHost
     private long _watchdogNextPass;
     private long _seq;
 
-    public WorkerHost(NamedPipeClientStream pipe, string nonce, int parentPid)
+    /// <summary>Test seam (<c>Hello.test_stall_reader_on_diag</c>).</summary>
+    private bool _testStallReaderOnDiag;
+
+    public WorkerHost(Stream pipe, string nonce, int parentPid)
     {
         _pipe = pipe;
         _nonce = nonce;
@@ -106,11 +110,10 @@ internal sealed class WorkerHost
     {
         try
         {
-            // Blocking I/O on Unix from the first byte (lmxopcua#789): one async call would turn the socket under the
-            // pipe non-blocking for good, and every later read would go through the socket engine.
-            var hello = ScriptWorkerFrameCodec.UseBlockingIo
-                ? ScriptWorkerFrameCodec.Read(_pipe, ScriptWorkerFrameCodec.DefaultMaxFrameBytes)
-                : await ScriptWorkerFrameCodec.ReadAsync(_pipe, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
+            // Blocking I/O from the first byte (lmxopcua#789): on Unix one async call would turn the socket under the
+            // pipe non-blocking for good, and every later read would go through the socket engine; on Windows the
+            // stream waits for its own overlapped I/O, off the completion port.
+            var hello = ScriptWorkerFrameCodec.Read(_pipe, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
             if (hello is null) return ScriptWorkerEnvironment.ExitPipeClosed;
             if (hello.BodyCase != Envelope.BodyOneofCase.Hello)
                 throw new ScriptWorkerProtocolException($"expected Hello, got {hello.BodyCase}");
@@ -128,14 +131,13 @@ internal sealed class WorkerHost
                     WorkerSoftClock = _workerSoftClock,
                 },
             });
-            if (ScriptWorkerFrameCodec.UseBlockingIo) ScriptWorkerFrameCodec.Write(_pipe, ready, _maxFrameBytes);
-            else await ScriptWorkerFrameCodec.WriteAsync(_pipe, ready, _maxFrameBytes);
+            ScriptWorkerFrameCodec.Write(_pipe, ready, _maxFrameBytes);
 
             _logSink.Capacity = (int)Math.Max(1, hello.Hello.MaxLogLines == 0 ? 1000 : hello.Hello.MaxLogLines);
             _writer = new ScriptWorkerFrameWriter(_pipe, OnWriteFault, _maxFrameBytes, firstSeq: Interlocked.Read(ref _seq),
-                synchronous: ScriptWorkerFrameCodec.UseBlockingIo);
+                synchronous: true);
             StartBackground(Math.Max(1, (int)hello.Hello.EvalThreads));
-            var exit = ScriptWorkerFrameCodec.UseBlockingIo ? await ReadOnDedicatedThread() : await ReadLoopAsync();
+            var exit = await ReadOnDedicatedThread();
             _writer.Complete();
             return exit;
         }
@@ -158,6 +160,7 @@ internal sealed class WorkerHost
         if (hello.HeartbeatIntervalMs > 0) _heartbeatInterval = TimeSpan.FromMilliseconds(hello.HeartbeatIntervalMs);
         if (hello.CompileThreads > 0) _compileThreads = (int)Math.Min(hello.CompileThreads, 64);
         _workerSoftClock = hello.WorkerSoftClock;
+        _testStallReaderOnDiag = hello.TestStallReaderOnDiag;
         if (!string.IsNullOrEmpty(hello.PeCacheDir) && hello.PeCacheKey.Length == ScriptPeCacheStore.KeyLength)
         {
             // Design §7: the host owns the key; it reaches us only here, never through the environment or a file.
@@ -181,7 +184,7 @@ internal sealed class WorkerHost
         if (_parentPid > 0) _ = Task.Run(ParentWatchLoopAsync);
     }
 
-    /// <summary>The blocking read loop on its own thread (Unix, lmxopcua#789); completes with the exit code.</summary>
+    /// <summary>The blocking read loop on its own thread (lmxopcua#789); completes with the exit code.</summary>
     private Task<int> ReadOnDedicatedThread()
     {
         var done = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -206,9 +209,21 @@ internal sealed class WorkerHost
                     done.TrySetResult(ScriptWorkerEnvironment.ExitPipeClosed);
                     return;
                 }
-                if (HandleFrame(frame) is { } exit)
+                int? exit;
+                try
                 {
-                    done.TrySetResult(exit);
+                    exit = HandleFrame(frame);
+                }
+                catch (Exception ex)
+                {
+                    // Never an unhandled exception on this thread: the worker ends as a fault the host can attribute.
+                    Console.Error.WriteLine($"ScriptWorker: reader failed: {ex}");
+                    done.TrySetResult(ScriptWorkerEnvironment.ExitProtocolFault);
+                    return;
+                }
+                if (exit is { } code)
+                {
+                    done.TrySetResult(code);
                     return;
                 }
             }
@@ -216,29 +231,6 @@ internal sealed class WorkerHost
         return done.Task;
     }
 
-    private async Task<int> ReadLoopAsync()
-    {
-        var reader = new ScriptWorkerFrameReader(_pipe, _maxFrameBytes);
-        while (true)
-        {
-            Envelope? frame;
-            try
-            {
-                frame = await reader.ReadAsync();
-            }
-            catch (ScriptWorkerProtocolException ex)
-            {
-                await Console.Error.WriteLineAsync($"ScriptWorker: protocol fault: {ex.Message}");
-                return ScriptWorkerEnvironment.ExitProtocolFault;
-            }
-            catch (IOException)
-            {
-                return ScriptWorkerEnvironment.ExitPipeClosed;
-            }
-            if (HandleFrame(frame) is { } exit) return exit;
-        }
-    }
-
     /// <summary>Acts on one frame from the host.</summary>
     /// <param name="frame">The frame; null = end of stream.</param>
     /// <returns>The exit code when the worker must end, otherwise null.</returns>
@@ -274,10 +266,18 @@ internal sealed class WorkerHost
             case Envelope.BodyOneofCase.Cancel:
                 if (_running.TryGetValue((long)frame.Cancel.EvalId, out var toCancel))
                 {
-                    try { toCancel.Cancellation.Cancel(); } catch (ObjectDisposedException) { /* just finished */ }
+                    // Off the reader (lmxopcua#789): Cancel() runs the script's own ctx.CancellationToken callbacks, and a
+                // callback that blocks would stop this worker reading its pipe — the host's writes would then block.
+                ThreadPool.UnsafeQueueUserWorkItem(static cts =>
+                {
+                    try { cts.Cancel(); }
+                    catch (ObjectDisposedException) { /* just finished */ }
+                    catch (Exception ex) { Console.Error.WriteLine($"ScriptWorker: a cancellation callback threw: {ex.Message}"); }
+                }, toCancel.Cancellation, preferLocal: false);
                 }
                 return null;
             case Envelope.BodyOneofCase.DiagRequest:
+                if (_testStallReaderOnDiag) Thread.Sleep(Timeout.Infinite); // test seam: talk but never listen again
                 Send(new Envelope { DiagReply = Diagnostics() });
                 return null;
             case Envelope.BodyOneofCase.Shutdown:

tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/BlockingPipeStreamTests.cs

@@ -0,0 +1,229 @@
+using System.IO.Pipes;
+using System.Security.Cryptography;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
+using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
+
+namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
+
+/// <summary>
+///     lmxopcua#789: <see cref="ScriptWorkerFrameCodec.BlockingStream"/> — on Windows the
+///     <see cref="WindowsBlockingPipeStream"/> that keeps worker pipe I/O off the completion port. Exercised on a real
+///     pipe pair; on Unix the same tests cover the plain pipe it returns there.
+/// </summary>
+[Trait("Category", "ScriptWorker")]
+public sealed class BlockingPipeStreamTests
+{
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+    private static async Task<(Stream Server, Stream Client)> PairAsync()
+    {
+        var name = "otopcua-sw-test-" + Convert.ToHexString(RandomNumberGenerator.GetBytes(6));
+        var server = new NamedPipeServerStream(name, PipeDirection.InOut, 1, PipeTransmissionMode.Byte,
+            PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
+        var client = new NamedPipeClientStream(".", name, PipeDirection.InOut, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
+        var accept = server.WaitForConnectionAsync(Ct);
+        client.Connect(TimeSpan.FromSeconds(5));
+        await accept;
+        return (ScriptWorkerFrameCodec.BlockingStream(server), ScriptWorkerFrameCodec.BlockingStream(client));
+    }
+
+    private static Envelope Frame(ulong id) => new()
+    {
+        ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
+        Seq = id,
+        Heartbeat = new Heartbeat { Inflight = (uint)id },
+    };
+
+    [Fact]
+    public async Task On_windows_the_pipe_is_wrapped_and_elsewhere_it_is_the_pipe_itself()
+    {
+        var (server, client) = await PairAsync();
+        using (server)
+        using (client)
+        {
+            if (OperatingSystem.IsWindows()) server.ShouldBeOfType<WindowsBlockingPipeStream>();
+            else server.ShouldBeAssignableTo<PipeStream>();
+        }
+    }
+
+    /// <summary>
+    ///     Full duplex: each end has a read blocked on a dedicated thread while the other direction keeps writing — a
+    ///     handle opened for synchronous I/O would serialise the write behind the blocked read and deadlock here.
+    /// </summary>
+    [Fact]
+    public async Task Reads_blocked_on_both_ends_do_not_hold_up_writes_in_either_direction()
+    {
+        var (server, client) = await PairAsync();
+        using (server)
+        using (client)
+        {
+            const int frames = 2_000;
+            Task<int> Pump(Stream from, Stream to)
+            {
+                var reader = Task.Factory.StartNew(() =>
+                {
+                    var r = new ScriptWorkerFrameReader(from);
+                    var seen = 0;
+                    while (seen < frames && r.Read() is { } f)
+                    {
+                        f.Heartbeat.Inflight.ShouldBe((uint)seen);
+                        seen++;
+                    }
+                    return seen;
+                }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);
+                _ = Task.Factory.StartNew(() =>
+                {
+                    for (var i = 0; i < frames; i++) ScriptWorkerFrameCodec.Write(to, Frame((ulong)i), ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
+                }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);
+                return reader;
+            }
+
+            var toClient = Pump(client, server);
+            var toServer = Pump(server, client);
+            (await toClient.WaitAsync(TimeSpan.FromSeconds(30), Ct)).ShouldBe(frames);
+            (await toServer.WaitAsync(TimeSpan.FromSeconds(30), Ct)).ShouldBe(frames);
+        }
+    }
+
+    [Fact]
+    public async Task A_large_frame_crosses_whole()
+    {
+        var (server, client) = await PairAsync();
+        using (server)
+        using (client)
+        {
+            var big = new Envelope
+            {
+                ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
+                Load = new LoadScripts { Scripts = { new ScriptSource { Key = "k", Source = new string('x', 1_000_000) } } },
+            };
+            var read = Task.Factory.StartNew(() => new ScriptWorkerFrameReader(server).Read(), Ct, TaskCreationOptions.LongRunning,
+                TaskScheduler.Default);
+            ScriptWorkerFrameCodec.Write(client, big, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
+            var got = await read.WaitAsync(TimeSpan.FromSeconds(30), Ct);
+            got!.Load.Scripts[0].Source.Length.ShouldBe(1_000_000);
+        }
+    }
+
+    [Fact]
+    public async Task A_peer_that_goes_away_ends_a_blocked_read_and_fails_a_write()
+    {
+        var (server, client) = await PairAsync();
+        using (server)
+        {
+            var read = Task.Factory.StartNew(() => server.Read(new byte[16], 0, 16), Ct, TaskCreationOptions.LongRunning,
+                TaskScheduler.Default);
+            await Task.Delay(100, Ct);
+            read.IsCompleted.ShouldBeFalse("positive control: the read is blocked on the kernel");
+            client.Dispose();
+            (await read.WaitAsync(TimeSpan.FromSeconds(10), Ct)).ShouldBe(0, "end of stream");
+            Should.Throw<IOException>(() =>
+            {
+                // A pipe can accept a write into its buffer before it notices; the second one must fail.
+                for (var i = 0; i < 4; i++) server.Write(new byte[64], 0, 64);
+            });
+        }
+    }
+
+    [Fact]
+    public async Task Disposing_an_end_wakes_its_own_blocked_reader()
+    {
+        var (server, client) = await PairAsync();
+        using (client)
+        {
+            var read = Task.Factory.StartNew(() => server.Read(new byte[16], 0, 16), Ct, TaskCreationOptions.LongRunning,
+                TaskScheduler.Default);
+            await Task.Delay(100, Ct);
+            read.IsCompleted.ShouldBeFalse("positive control: the read is blocked on the kernel");
+            server.Dispose();
+            // End of stream, or the disposed / I/O error the readers treat as the pipe ending — never a hang.
+            int ended;
+            try { ended = await read.WaitAsync(TimeSpan.FromSeconds(10), Ct); }
+            catch (Exception ex) when (ex is ObjectDisposedException or IOException) { ended = 0; }
+            ended.ShouldBe(0);
+        }
+    }
+
+    /// <summary>
+    ///     Writes until the pipe is full (the peer never reads), then returns the writing task, which is blocked in the
+    ///     kernel at that point; it ends with the exception the write saw.
+    /// </summary>
+    private static async Task<Task> BlockedWriterAsync(Stream end)
+    {
+        long written = 0;
+        var writer = Task.Factory.StartNew(() =>
+        {
+            var chunk = new byte[64 * 1024];
+            while (true)
+            {
+                end.Write(chunk, 0, chunk.Length);
+                Interlocked.Add(ref written, chunk.Length);
+            }
+        }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);
+        // Full once no progress is made for a while (a zero-quota Windows pipe blocks at once; a socket after its buffer).
+        long last;
+        do
+        {
+            last = Interlocked.Read(ref written);
+            await Task.Delay(300, Ct);
+        } while (Interlocked.Read(ref written) != last && !writer.IsCompleted);
+        writer.IsCompleted.ShouldBeFalse("positive control: the write is blocked on a full pipe");
+        return writer;
+    }
+
+    private static async Task ShouldEndAsync(Task blocked, string what)
+    {
+        var ended = await Task.WhenAny(blocked, Task.Delay(TimeSpan.FromSeconds(10), Ct));
+        ended.ShouldBeSameAs(blocked, $"{what} ended, not hung");
+        var ex = blocked.Exception?.GetBaseException();
+        ex.ShouldNotBeNull($"{what} ended with an error");
+        (ex is ObjectDisposedException or IOException).ShouldBeTrue(ex.ToString());
+    }
+
+    [Fact]
+    public async Task Disposing_an_end_wakes_its_own_write_blocked_on_a_full_pipe()
+    {
+        var (server, client) = await PairAsync();
+        using (client)
+        {
+            var writer = await BlockedWriterAsync(server);
+            server.Dispose();
+            await ShouldEndAsync(writer, "the blocked write");
+        }
+    }
+
+    [Fact]
+    public async Task Disposing_an_end_with_a_read_and_a_write_both_pending_wakes_both()
+    {
+        var (server, client) = await PairAsync();
+        using (client)
+        {
+            var read = Task.Factory.StartNew(() =>
+            {
+                var n = server.Read(new byte[16], 0, 16);
+                if (n == 0) throw new IOException("end of stream");
+                return n;
+            }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);
+            var writer = await BlockedWriterAsync(server);
+            read.IsCompleted.ShouldBeFalse("positive control: the read is pending beside the blocked write");
+            server.Dispose();
+            await ShouldEndAsync(writer, "the blocked write");
+            await ShouldEndAsync(read, "the pending read");
+        }
+    }
+
+    [Fact]
+    public async Task The_asynchronous_surface_is_refused_on_windows()
+    {
+        var (server, client) = await PairAsync();
+        using (server)
+        using (client)
+        {
+            if (!OperatingSystem.IsWindows()) return; // the Unix pipe itself keeps PipeStream's own async surface
+            await Should.ThrowAsync<NotSupportedException>(() => server.ReadAsync(new byte[4], 0, 4, Ct));
+            await Should.ThrowAsync<NotSupportedException>(async () => await server.WriteAsync(new byte[4].AsMemory(), Ct));
+        }
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/HandshakeTests.cs

@@ -9,6 +9,7 @@ using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
 namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
 
 /// <summary>lmxopcua#780 design §6 "Pipe hardening": only the launched process, echoing its nonce, is accepted.</summary>
+[Trait("Category", "ScriptWorker")]
 public sealed class HandshakeTests
 {
     private static CancellationToken Ct => TestContext.Current.CancellationToken;

tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/PipeFaultTests.cs

@@ -0,0 +1,106 @@
+using System.Diagnostics;
+using System.Security.Cryptography;
+using System.Text;
+using Serilog;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Scripting;
+using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
+using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
+
+namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
+
+/// <summary>
+///     lmxopcua#789 review: the blocking pipe I/O runs the host's reader on a dedicated thread and blocks producers on a
+///     full pipe. A fault in the reader's inline completion work must end that worker, not the server process; and a worker
+///     that stops reading while it keeps heartbeating must be killed so blocked producers are released.
+/// </summary>
+[Collection(ScriptWorkerCollection.Name)]
+[Trait("Category", "ScriptWorker")]
+public sealed class PipeFaultTests
+{
+    private const string Healthy = "return 7;";
+    private const string Runaway = "while (true) { }";
+
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+    /// <summary>
+    ///     The slot release runs on the pipe reader, inline, before the caller resumes. When it throws, the reader must
+    ///     end the worker as a fault: the run in flight beside it answers BadResourceUnavailable ("worker restarted"),
+    ///     the shard restarts, and the next evaluation is served — instead of an unhandled exception on a dedicated thread,
+    ///     which would terminate the host.
+    /// </summary>
+    [Fact]
+    public async Task A_completion_callback_that_throws_on_the_reader_ends_that_worker_as_faulted_not_the_host()
+    {
+        await using var runner = Scripts.Runner(Scripts.Options());
+        (await runner.RunAsync(Scripts.Vt("boom", Healthy), Ct)).Value.ShouldBe(7, "positive control: the hook is not armed yet");
+        var worker = runner.Shards[0].Current.ShouldNotBeNull();
+
+        var spin = runner.RunAsync(Scripts.Vt("spin", Runaway, TimeSpan.FromSeconds(30)), Ct);
+        await Scripts.Eventually(() => worker.InFlightCount > 0 ? worker : null, TimeSpan.FromSeconds(30), "the sibling in flight");
+
+        var armed = 1;
+        runner.SlotReleasingForTest = key =>
+        {
+            if (key.EndsWith("boom", StringComparison.Ordinal) && Interlocked.Exchange(ref armed, 0) == 1)
+                throw new InvalidOperationException("a completion callback failed");
+        };
+        await runner.RunAsync(Scripts.Vt("boom", Healthy), Ct);
+
+        var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(15), Ct);
+        exit.Reason.ShouldBe(WorkerExitReason.ProtocolFault);
+        var sibling = await spin.WaitAsync(TimeSpan.FromSeconds(15), Ct);
+        sibling.Status.ShouldBe(ScriptRunStatus.Unavailable);
+        sibling.Reason.ShouldBe(ScriptRunOutcome.WorkerRestartedReason);
+        sibling.ToVirtualTagEvalResult().StatusCode.ShouldBe(ZB.MOM.WW.OtOpcUa.Commons.OpcUa.OpcUaStatus.BadResourceUnavailable);
+
+        (await runner.RunAsync(Scripts.Vt("after", Healthy), Ct)).Value.ShouldBe(7, "the shard restarted and serves again");
+        runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldNotBe(worker.Pid);
+    }
+
+    /// <summary>
+    ///     A worker whose reader stops while its heartbeats go on: the host's writes fill the pipe and the producer (here
+    ///     a deploy loading scripts, the case nothing else would end: no evaluation is outstanding and a compile's timeout
+    ///     starts only when the worker starts it) blocks in the write. The write-stall check must kill the worker within
+    ///     its bound — as unresponsive, not for silence, which proves it kept heartbeating — and that releases the producer.
+    /// </summary>
+    [Fact]
+    public async Task A_worker_that_stops_reading_but_keeps_heartbeating_is_killed_and_the_blocked_producer_returns()
+    {
+        var heartbeat = TimeSpan.FromMilliseconds(200);
+        var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }),
+            256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: heartbeat,
+            HandshakeTimeout: TimeSpan.FromSeconds(15), HeartbeatSilenceLimit: TimeSpan.FromSeconds(30), TestStallReaderOnDiag: true);
+        var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct);
+        await using var owned = connection;
+        var warm = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Healthy,
+            SHA256.HashData(Encoding.UTF8.GetBytes(Healthy)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct);
+        warm.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue();
+        connection.SilenceLimit = heartbeat * 5; // armed once the cold start is over (see SupervisionTests)
+        var stallLimit = TimeSpan.FromSeconds(2);
+        connection.WriteStallLimitOverride = stallLimit;
+
+        _ = connection.DiagnosticsAsync(TimeSpan.FromSeconds(60)); // the worker's reader stops here for good
+        var producer = Task.Factory.StartNew(() =>
+        {
+            // About 5 MB of loads: far more than the pipe's buffer and the worker reader's own.
+            for (var i = 0; i < 20; i++)
+            {
+                var source = $"return {i}; // " + new string('x', 256 * 1024);
+                _ = connection.EnsureLoadedAsync($"big{i}", ContextKind.VirtualTag, source,
+                    SHA256.HashData(Encoding.UTF8.GetBytes(source)), []);
+            }
+        }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);
+
+        var sw = Stopwatch.StartNew();
+        await Task.Delay(stallLimit / 2, Ct);
+        producer.IsCompleted.ShouldBeFalse("positive control: the producer is blocked in a write to the full pipe");
+        connection.IsAlive.ShouldBeTrue("positive control: heartbeats keep the silence check quiet");
+
+        var exit = await connection.Exited.WaitAsync(TimeSpan.FromSeconds(20), Ct);
+        exit.Reason.ShouldBe(WorkerExitReason.KilledUnresponsive, "killed by the write-stall check, not for silence");
+        await producer.WaitAsync(TimeSpan.FromSeconds(10), Ct);
+        sw.Elapsed.ShouldBeLessThan(stallLimit + TimeSpan.FromSeconds(8), "the producer was released within the bound");
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ProtocolTests.cs

@@ -9,6 +9,7 @@ using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
 namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
 
 /// <summary>lmxopcua#780 design §13 — frame codec and the CLR ↔ Variant mapping.</summary>
+[Trait("Category", "ScriptWorker")]
 public sealed class ProtocolTests
 {
     public static IEnumerable<object?[]> Values()