fix(worker-tests): model the STA pump at heartbeat capture — WorkerPipeSessionTests long-in-flight repro

RunAsync_LongInFlightCommandThatKeepsPumping_DoesNotFaultAndDeliversReply failed
deterministically on the Windows box with a StaHung fault whose command_method was
empty and whose staleness was 233 ms — i.e. a fault raised with no command in flight,
before the scenario under test began. FakeRuntimeSession stamps LastStaActivityUtc once,
at construction, and the test only started refreshing it after the blocked dispatch
signalled. Everything between those two points — handshake, STA init, the first
heartbeat — captured a snapshot already stale past the compressed 50 ms grace, with no
correlation id for the watchdog to suppress on, so the watchdog correctly reported the
fake as hung.

The harness, not the product, was wrong: StaRuntime.ThreadMain calls MarkActivity() on
every WaitForWorkOrMessages iteration, so a live worker is never captured stale, idle or
busy. Model that where it belongs — FakeRuntimeSession.RefreshStaActivityOnCapture (opt
in, default off) stamps activity at each CaptureHeartbeat and leaves the rest of the
snapshot alone — and arm it before RunAsync so the first beat is covered. The test-owned
refresh loop goes away with it; a thread-pool loop racing a compressed grace could not
have held the invariant anyway.

Scenario intent is unchanged and slightly stronger: the command still blocks in dispatch
across 30 heartbeats (~600 ms, many multiples of the 100 ms stuck ceiling), no frame may
be a fault, and the reply must still arrive. The reply leg is now fault-checked too
(previously it skipped frames blindly), and the pump keeps running across the release, as
it does in production while the reply is marshalled off the STA. Fault assertions now
report the category and diagnostic message instead of a bare body-case mismatch.

Test-only change; no product code, frame protocol, or STA rule touched.

src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs

@@ -1453,8 +1453,15 @@ public sealed class WorkerPipeSessionTests
     ///     <c>HeartbeatStuckCeiling</c> (75 s in production) keeps its activity
     ///     timestamp fresh. This test compresses the clock — a 100 ms ceiling
     ///     with a command in flight across a window many multiples longer — and
-    ///     models the pump refresh by continuously advancing the snapshot's
-    ///     <c>LastStaActivityUtc</c> while the command blocks. Contrast
+    ///     models the pump refresh with
+    ///     <see cref="FakeRuntimeSession.RefreshStaActivityOnCapture"/>, which
+    ///     stamps activity at every heartbeat capture exactly as the pump's
+    ///     per-iteration <c>MarkActivity()</c> does. The refresh has to be in
+    ///     effect from construction, not from the moment the command blocks:
+    ///     with a 50 ms grace, the idle window covering handshake and startup
+    ///     carries no correlation id for the watchdog to suppress on, so a fake
+    ///     whose activity timestamp is frozen at construction is reported
+    ///     <c>StaHung</c> before the scenario under test even starts. Contrast
     ///     <see cref="RunAsync_WhenStaActivityIsStaleBeyondCeilingWithCommandInFlight_WritesWatchdogFault"/>,
     ///     where a frozen timestamp beyond the ceiling correctly faults; here
     ///     the refreshed timestamp must keep the fault suppressed and let the
@@ -1469,6 +1476,13 @@ public sealed class WorkerPipeSessionTests
         FakeRuntimeSession runtime = new()
         {
             BlockDispatch = true,
+
+            // The pump refreshes STA activity on every wait iteration, so every
+            // heartbeat capture on a healthy worker sees fresh activity — while
+            // a command holds the STA and while it is idle alike. Armed before
+            // RunAsync so the very first beat, sent as soon as the session is
+            // Ready, is already covered.
+            RefreshStaActivityOnCapture = true,
         };
         WorkerPipeSession session = CreatePipeSession(
             pipePair.WorkerStream,
@@ -1490,49 +1504,48 @@ public sealed class WorkerPipeSessionTests
             runtime.DispatchStarted.Wait(TimeSpan.FromSeconds(5)),
             "The long command must reach the runtime and begin dispatch.");
 
-        // Model the pump refreshing STA activity on each wait iteration: keep
-        // the snapshot's LastStaActivityUtc current while the command is in
-        // flight.
-        using CancellationTokenSource pumpRefresh = new();
-        Task refreshLoop = Task.Run(
-            async () =>
-            {
-                while (!pumpRefresh.IsCancellationRequested)
-                {
-                    runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
-                        DateTimeOffset.UtcNow,
-                        pendingCommandCount: 1,
-                        outboundEventQueueDepth: 0,
-                        lastEventSequence: 0,
-                        currentCommandCorrelationId: "long-bulk-read"));
-                    await Task.Delay(TimeSpan.FromMilliseconds(20)).ConfigureAwait(false);
-                }
-            });
+        // Publish the in-flight shape the heartbeat then reports for the whole
+        // blocked window; only LastStaActivityUtc moves after this, refreshed by
+        // the modelled pump at each capture.
+        runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
+            DateTimeOffset.UtcNow,
+            pendingCommandCount: 1,
+            outboundEventQueueDepth: 0,
+            lastEventSequence: 0,
+            currentCommandCorrelationId: "long-bulk-read"));
 
         // Inspect a bounded number of frames over a window many multiples of the
         // 100 ms ceiling (at least 30 heartbeats at 20 ms ~ 600 ms). None may be
         // a WorkerFault while activity is continuously refreshed.
         const int framesToInspect = 30;
-        for (int index = 0; index < framesToInspect; index++)
+        int frameIndex = 0;
+        for (; frameIndex < framesToInspect; frameIndex++)
         {
             WorkerEnvelope envelope = await pipePair.GatewayReader
                 .ReadAsync(cancellation.Token);
-            Assert.NotEqual(
-                WorkerEnvelope.BodyOneofCase.WorkerFault,
-                envelope.BodyCase);
+            AssertNotFault(envelope, frameIndex);
         }
 
-        // Stop refreshing and release the command; its reply must be delivered
-        // because the session never faulted (state stayed Ready).
-        pumpRefresh.Cancel();
-        await refreshLoop;
+        // Release the command with the pump still running — as it is in
+        // production while the reply is marshalled off the STA. The reply must
+        // be delivered (the session never faulted, so its state stayed Ready),
+        // and no frame on the way to it may be a fault either.
         runtime.ReleaseDispatch();
 
-        WorkerEnvelope reply = await ReadUntilAsync(
-            pipePair.GatewayReader,
-            WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
-            envelope => envelope.CorrelationId == "long-bulk-read",
-            cancellation.Token);
+        WorkerEnvelope reply;
+        while (true)
+        {
+            WorkerEnvelope envelope = await pipePair.GatewayReader
+                .ReadAsync(cancellation.Token);
+            AssertNotFault(envelope, frameIndex++);
+            if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerCommandReply
+                && envelope.CorrelationId == "long-bulk-read")
+            {
+                reply = envelope;
+                break;
+            }
+        }
+
         Assert.Equal(
             ProtocolStatusCode.Ok,
             reply.WorkerCommandReply.Reply.ProtocolStatus.Code);
@@ -2217,6 +2230,26 @@ public sealed class WorkerPipeSessionTests
     /// <param name="predicate">Predicate to match against envelope.</param>
     /// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
     /// <returns>The matching envelope.</returns>
+    /// <summary>
+    ///     Fails when the frame is a <c>WorkerFault</c>, naming the category and diagnostic message.
+    ///     A bare body-case comparison reports only "expected not WorkerFault", which says nothing
+    ///     about which watchdog or protocol path produced it — the one fact needed to tell a
+    ///     regression from a harness that mis-models the runtime.
+    /// </summary>
+    /// <param name="envelope">Frame read from the gateway end.</param>
+    /// <param name="frameIndex">Ordinal of the frame within the inspected run.</param>
+    private static void AssertNotFault(WorkerEnvelope envelope, int frameIndex)
+    {
+        if (envelope.BodyCase != WorkerEnvelope.BodyOneofCase.WorkerFault)
+        {
+            return;
+        }
+
+        Assert.Fail(
+            $"Frame {frameIndex} is a WorkerFault ({envelope.WorkerFault.Category}): "
+            + envelope.WorkerFault.DiagnosticMessage);
+    }
+
     private static async Task<WorkerEnvelope> ReadUntilAsync(
         WorkerFrameReader reader,
         WorkerEnvelope.BodyOneofCase expectedBody,

src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs

@@ -127,11 +127,54 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
             });
     }
 
+    private bool refreshStaActivityOnCapture;
+
+    /// <summary>
+    ///     When set, <see cref="CaptureHeartbeat"/> stamps the snapshot's
+    ///     <c>LastStaActivityUtc</c> with the capture time and leaves every other field as the last
+    ///     <see cref="SetSnapshot"/> left it. Models a live STA whose pump calls
+    ///     <c>MarkActivity()</c> on each wait iteration (<c>StaRuntime.ThreadMain</c>), so a healthy
+    ///     worker is never captured stale — which a watchdog test needs to hold for the <em>whole</em>
+    ///     session, including the handshake window before any command exists for the watchdog to
+    ///     suppress on. A test-owned refresh loop cannot hold it: it is a thread-pool continuation
+    ///     racing a compressed grace, and the gap between this fake being constructed and that loop's
+    ///     first tick is already enough to look hung.
+    /// </summary>
+    public bool RefreshStaActivityOnCapture
+    {
+        get
+        {
+            lock (gate)
+            {
+                return refreshStaActivityOnCapture;
+            }
+        }
+
+        set
+        {
+            lock (gate)
+            {
+                refreshStaActivityOnCapture = value;
+            }
+        }
+    }
+
     /// <inheritdoc />
     public WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat()
     {
         lock (gate)
         {
+            if (refreshStaActivityOnCapture)
+            {
+                snapshot = new WorkerRuntimeHeartbeatSnapshot(
+                    DateTimeOffset.UtcNow,
+                    snapshot.PendingCommandCount,
+                    snapshot.OutboundEventQueueDepth,
+                    snapshot.LastEventSequence,
+                    snapshot.CurrentCommandCorrelationId,
+                    snapshot.StaCallInProgress);
+            }
+
             return snapshot;
         }
     }