fix: address final ackmsg-as-operator review findings

- AlarmAckWriter: classify AuthenticateUser and AddItem/Advise/Write
  failures by HRESULT for any interop exception (ArgumentException for
  E_INVALIDARG, UnauthorizedAccessException, COMException); server-gone
  HRESULTs at AuthenticateUser tear down the registration and report
  WriteRejected.
- Bound the ack item cache with a 256-entry LRU (eviction best-effort
  UnAdvise+RemoveItem).
- After an Unconfirmed ack, drop and re-add the item before the next
  write so a late OnWriteComplete on the old handle is not credited.
- Pass StaRuntime.PumpPendingMessages to the ack writer through a
  pump-aware alarm handler factory so long completion waits refresh the
  STA watchdog heartbeat.
- Rust CLI: --operator is required for acknowledge-alarm.
- Docs: full ack status table, credential-oracle trade-off, upgrade note.

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

src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmAckWriterTests.cs

@@ -84,7 +84,9 @@ public sealed class AlarmAckWriterTests
     public void Acknowledge_Twice_ReusesItemHandle()
     {
         RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
-        using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
+        MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
+        server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
+        using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
 
         writer.Acknowledge("T.A.AckMsg", "one", Alice());
         writer.Acknowledge("T.A.AckMsg", "two", Alice());
@@ -114,7 +116,9 @@ public sealed class AlarmAckWriterTests
     public void Acknowledge_WhenCachedWriteThrows_RejectsAndReAddsNextTime()
     {
         RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
-        using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
+        MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
+        server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
+        using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
         writer.Acknowledge("T.A.AckMsg", "one", Alice());
         server.WriteException = new COMException("x", unchecked((int)0x80004005));
 
@@ -141,6 +145,130 @@ public sealed class AlarmAckWriterTests
         Assert.Contains("Unregister", server.Operations);
     }
 
+    [Fact]
+    public void Acknowledge_WhenAuthenticateUserDisconnected_TearsDownAndRejects()
+    {
+        RecordingAckServer server = new RecordingAckServer
+        {
+            AuthenticateResult = 5,
+            AuthenticateException = new COMException("gone pw-secret", unchecked((int)0x80010108)),
+        };
+        using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
+
+        AlarmAckResult result = writer.Acknowledge("T.A.AckMsg", "one", Alice());
+
+        Assert.Equal(AlarmAckOutcome.WriteRejected, result.Outcome);
+        Assert.Equal(unchecked((int)0x80010108), result.NativeStatus);
+        Assert.Contains("AuthenticateUser", result.Diagnostic, StringComparison.Ordinal);
+        Assert.Contains("0x80010108", result.Diagnostic, StringComparison.Ordinal);
+        Assert.DoesNotContain("pw-secret", result.Diagnostic, StringComparison.Ordinal);
+        Assert.Contains("Unregister", server.Operations);
+        Assert.DoesNotContain(server.Operations, op => op.StartsWith("AddItem", StringComparison.Ordinal));
+    }
+
+    [Fact]
+    public void Acknowledge_WhenAuthenticateUserThrowsArgumentException_ReturnsAuthenticationFailed()
+    {
+        // Live MXAccess raises E_INVALIDARG for a bad user or wrong password, which the
+        // interop surfaces as ArgumentException rather than COMException.
+        RecordingAckServer server = new RecordingAckServer
+        {
+            AuthenticateResult = 5,
+            AuthenticateException = new ArgumentException("Value does not fall within the expected range. pw-secret"),
+        };
+        using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
+
+        AlarmAckResult result = writer.Acknowledge("T.A.AckMsg", "one", Alice());
+
+        Assert.Equal(AlarmAckOutcome.AuthenticationFailed, result.Outcome);
+        Assert.Equal(unchecked((int)0x80070057), result.NativeStatus);
+        Assert.DoesNotContain("pw-secret", result.Diagnostic, StringComparison.Ordinal);
+        Assert.DoesNotContain(server.Operations, op => op.StartsWith("Write", StringComparison.Ordinal));
+        Assert.DoesNotContain("Unregister", server.Operations);
+    }
+
+    [Fact]
+    public void Acknowledge_WhenWriteThrowsArgumentException_RejectsAndDropsItem()
+    {
+        RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5, WriteException = new ArgumentException("bad") };
+        using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(10), "MxGateway");
+
+        AlarmAckResult result = writer.Acknowledge("T.A.AckMsg", "one", Alice());
+
+        Assert.Equal(AlarmAckOutcome.WriteRejected, result.Outcome);
+        Assert.Equal(unchecked((int)0x80070057), result.NativeStatus);
+        Assert.Contains("Write", result.Diagnostic, StringComparison.Ordinal);
+        Assert.Equal(new[] { 100 }, server.RemovedItemHandles);
+        Assert.DoesNotContain("Unregister", server.Operations);
+    }
+
+    [Fact]
+    public void Acknowledge_PastCacheCap_EvictsLeastRecentlyUsedItem()
+    {
+        RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
+        using AlarmAckWriter writer = new AlarmAckWriter(server, new MxAccessWriteCompletionCache(), () => { }, TimeSpan.FromMilliseconds(1), "MxGateway", maxCachedItems: 2);
+
+        writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100
+        writer.Acknowledge("B.X.AckMsg", "2", Alice()); // handle 101
+        writer.Acknowledge("C.X.AckMsg", "3", Alice()); // handle 102, evicts A (100)
+
+        Assert.Equal(new[] { 100 }, server.RemovedItemHandles);
+        Assert.Equal(3, server.Operations.Count(op => op.StartsWith("AddItem", StringComparison.Ordinal)));
+    }
+
+    [Fact]
+    public void Acknowledge_ReusedItem_IsMostRecentlyUsedForEviction()
+    {
+        RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
+        MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
+        server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
+        using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway", maxCachedItems: 2);
+
+        writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100
+        writer.Acknowledge("B.X.AckMsg", "2", Alice()); // handle 101
+        writer.Acknowledge("A.X.AckMsg", "3", Alice()); // reuse 100, now most recent
+        writer.Acknowledge("C.X.AckMsg", "4", Alice()); // handle 102, evicts B (101)
+
+        Assert.Equal(new[] { 101 }, server.RemovedItemHandles);
+        Assert.Equal(3, server.Operations.Count(op => op.StartsWith("AddItem", StringComparison.Ordinal)));
+        Assert.Contains("UnAdvise", server.Operations);
+    }
+
+    [Fact]
+    public void Acknowledge_AfterUnconfirmed_UsesFreshHandleSoStaleCompletionIsNotCredited()
+    {
+        RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
+        MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
+        using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromMilliseconds(50), "MxGateway");
+
+        AlarmAckResult first = writer.Acknowledge("T.A.AckMsg", "one", Alice()); // handle 100, no completion
+        // The first write's late OnWriteComplete (handle 100) lands while the
+        // second write is in flight; the second write's own completion never comes.
+        server.OnWrite = (handle, item) => cache.Record(handle, 100, Rows(MxStatusCategory.Ok, 0));
+        AlarmAckResult second = writer.Acknowledge("T.A.AckMsg", "two", Alice());
+
+        Assert.Equal(AlarmAckOutcome.Unconfirmed, first.Outcome);
+        Assert.Equal(AlarmAckOutcome.Unconfirmed, second.Outcome);
+        Assert.Contains(100, server.RemovedItemHandles);
+        Assert.Equal(2, server.Operations.Count(op => op.StartsWith("AddItem", StringComparison.Ordinal)));
+        Assert.Equal(new[] { 100, 101 }, server.WrittenItemHandles);
+    }
+
+    [Fact]
+    public void Acknowledge_AfterConfirmed_ReusesHandle()
+    {
+        RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 };
+        MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache();
+        server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0));
+        using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway");
+
+        writer.Acknowledge("T.A.AckMsg", "one", Alice());
+        writer.Acknowledge("T.A.AckMsg", "two", Alice());
+
+        Assert.Equal(new[] { 100, 100 }, server.WrittenItemHandles);
+        Assert.Empty(server.RemovedItemHandles);
+    }
+
     [Fact]
     public void Acknowledge_WhenNonOkCompletionHasNoDetail_ReturnsMinusOne()
     {
@@ -180,6 +308,10 @@ internal sealed class RecordingAckServer : IMxAccessServer
 
     public Action<int, int>? OnWrite { get; set; }
 
+    public List<int> RemovedItemHandles { get; } = new();
+
+    public List<int> WrittenItemHandles { get; } = new();
+
     public int Register(string clientName)
     {
         Operations.Add("Register:" + clientName);
@@ -196,7 +328,11 @@ internal sealed class RecordingAckServer : IMxAccessServer
 
     public int AddItem2(int serverHandle, string itemDefinition, string itemContext) => AddItem(serverHandle, itemDefinition);
 
-    public void RemoveItem(int serverHandle, int itemHandle) => Operations.Add("RemoveItem");
+    public void RemoveItem(int serverHandle, int itemHandle)
+    {
+        Operations.Add("RemoveItem");
+        RemovedItemHandles.Add(itemHandle);
+    }
 
     public void Advise(int serverHandle, int itemHandle)
     {
@@ -216,6 +352,7 @@ internal sealed class RecordingAckServer : IMxAccessServer
     public void Write(int serverHandle, int itemHandle, object? value, int userId)
     {
         Operations.Add("Write:" + value + ":" + userId);
+        WrittenItemHandles.Add(itemHandle);
         if (WriteException is not null)
         {
             throw WriteException;

src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs

@@ -76,7 +76,7 @@ public sealed class WorkerPipeSession
             options,
             () => Process.GetCurrentProcess().Id,
             new WorkerPipeSessionOptions(),
-            () => new MxAccessStaSession((eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null)),
+            () => new MxAccessStaSession((eq, affinity, comFactory, pump) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null, ackPumpStep: pump)),
             logger,
             stream)
     {
@@ -98,7 +98,7 @@ public sealed class WorkerPipeSession
             options,
             processIdProvider,
             new WorkerPipeSessionOptions(),
-            () => new MxAccessStaSession((eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null)),
+            () => new MxAccessStaSession((eq, affinity, comFactory, pump) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null, ackPumpStep: pump)),
             logger: null)
     {
     }
@@ -1585,7 +1585,7 @@ public sealed class WorkerPipeSession
         // parameterless CompleteStartupHandshakeAsync is used without a
         // prior factory call.
         _runtimeSession ??= new MxAccessStaSession(
-            (eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null));
+            (eq, affinity, comFactory, pump) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null, ackPumpStep: pump));
         IWorkerRuntimeSession session = _runtimeSession;
         try
         {

src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmAckWriter.cs

@@ -26,13 +26,25 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
     /// <summary>Default MXAccess client name when none is configured.</summary>
     public const string DefaultClientName = "MxGateway";
 
+    /// <summary>
+    ///     Default cap on cached ack item handles. Each cached item holds an
+    ///     advised MXAccess item for the worker's life, so the cache is bounded
+    ///     (least-recently-used eviction) rather than growing with every
+    ///     distinct alarm ever acknowledged.
+    /// </summary>
+    public const int DefaultMaxCachedItems = 256;
+
     private readonly IMxAccessComObjectFactory? factory;
     private readonly MxAccessWriteCompletionCache completionCache;
     private readonly Action pumpStep;
     private readonly TimeSpan completionTimeout;
     private readonly string clientName;
     private readonly MxStatusProxyConverter statusConverter = new MxStatusProxyConverter();
-    private readonly Dictionary<string, int> itemHandles = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
+    private readonly int maxCachedItems;
+    private readonly Dictionary<string, LinkedListNode<AckItem>> itemHandles = new Dictionary<string, LinkedListNode<AckItem>>(StringComparer.OrdinalIgnoreCase);
+
+    // Most-recently-used first; the tail is the eviction candidate.
+    private readonly LinkedList<AckItem> itemLru = new LinkedList<AckItem>();
     private object? comObject;
     private LMXProxyServerClass? comEventSource;
     private IMxAccessServer? server;
@@ -54,6 +66,7 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
         this.pumpStep = pumpStep ?? throw new ArgumentNullException(nameof(pumpStep));
         this.completionTimeout = completionTimeout;
         this.clientName = string.IsNullOrWhiteSpace(clientName) ? DefaultClientName : clientName!;
+        maxCachedItems = DefaultMaxCachedItems;
         completionCache = new MxAccessWriteCompletionCache();
     }
 
@@ -63,8 +76,11 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
         MxAccessWriteCompletionCache completionCache,
         Action pumpStep,
         TimeSpan completionTimeout,
-        string clientName)
+        string clientName,
+        int maxCachedItems = DefaultMaxCachedItems)
     {
+        if (maxCachedItems < 1) throw new ArgumentOutOfRangeException(nameof(maxCachedItems));
+        this.maxCachedItems = maxCachedItems;
         this.server = server ?? throw new ArgumentNullException(nameof(server));
         this.completionCache = completionCache ?? throw new ArgumentNullException(nameof(completionCache));
         this.pumpStep = pumpStep ?? throw new ArgumentNullException(nameof(pumpStep));
@@ -95,8 +111,22 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
         {
             userId = mx.AuthenticateUser(serverHandle, ackOperator.VerifyUser, ackOperator.Password ?? string.Empty);
         }
-        catch (COMException ex)
+        catch (Exception ex) when (IsInteropFailure(ex))
         {
+            // Bad credentials surface as ArgumentException (E_INVALIDARG) from the
+            // interop, not only COMException, so classify by the carried HRESULT.
+            int authHresult = ex.HResult;
+            if (IsServerGone(authHresult))
+            {
+                // The registration is dead, not the credentials: tear down so the
+                // next ack re-registers, and report it as a write-path failure.
+                TearDownServer();
+                return AlarmAckResult.Create(
+                    AlarmAckOutcome.WriteRejected,
+                    authHresult,
+                    $"AckMsg AuthenticateUser failed (HRESULT 0x{authHresult:X8}).");
+            }
+
             return AlarmAckResult.Create(
                 AlarmAckOutcome.AuthenticationFailed,
                 ex.HResult,
@@ -113,14 +143,32 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
 
         ulong baselineVersion = 0;
         int itemHandle = 0;
-        bool cached = itemHandles.TryGetValue(itemAddress, out itemHandle);
+        LinkedListNode<AckItem>? node;
+        bool cached = itemHandles.TryGetValue(itemAddress, out node);
         string step = "AddItem";
         try
         {
-            if (!cached)
+            if (cached && node!.Value.Dirty)
+            {
+                // The previous ack to this item ended Unconfirmed, so its
+                // OnWriteComplete may still arrive. Completions are keyed by
+                // (server, item) handle with no per-write id, so a late one would
+                // be credited to this write. Drop the item and re-add it so this
+                // write waits on a fresh handle the stale completion cannot match.
+                DropItem(mx, node);
+                cached = false;
+            }
+
+            if (cached)
+            {
+                itemHandle = node!.Value.Handle;
+                itemLru.Remove(node);
+                itemLru.AddFirst(node);
+            }
+            else
             {
                 itemHandle = mx.AddItem(serverHandle, itemAddress);
-                itemHandles[itemAddress] = itemHandle;
+                node = CacheItem(mx, itemAddress, itemHandle);
                 step = "Advise";
                 // MXAccess rejects writes to added-but-unadvised items with E_INVALIDARG.
                 mx.Advise(serverHandle, itemHandle);
@@ -131,13 +179,15 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
             baselineVersion = completionCache.CurrentVersion(serverHandle, itemHandle);
             mx.Write(serverHandle, itemHandle, comment ?? string.Empty, userId);
         }
-        catch (COMException ex)
+        catch (Exception ex) when (IsInteropFailure(ex))
         {
+            // The interop maps E_INVALIDARG to ArgumentException (and
+            // E_ACCESSDENIED to UnauthorizedAccessException), so the HRESULT, not
+            // the exception type, is what identifies the MXAccess failure.
             int hresult = ex.HResult;
-            if (itemHandles.Remove(itemAddress))
+            if (itemHandles.TryGetValue(itemAddress, out LinkedListNode<AckItem>? failed))
             {
-                try { mx.UnAdvise(serverHandle, itemHandle); } catch { /* best effort */ }
-                try { mx.RemoveItem(serverHandle, itemHandle); } catch { /* best effort */ }
+                DropItem(mx, failed!);
             }
 
             if (IsServerGone(hresult))
@@ -162,6 +212,7 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
                 pumpStep,
                 out RepeatedField<MxStatusProxy> statuses))
         {
+            node!.Value.Dirty = true;
             return AlarmAckResult.Create(
                 AlarmAckOutcome.Unconfirmed,
                 0,
@@ -192,6 +243,17 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
         TearDownServer();
     }
 
+    /// <summary>
+    ///     Exceptions the MXAccess COM interop raises for a failed HRESULT: the
+    ///     runtime maps well-known HRESULTs to typed exceptions (E_INVALIDARG to
+    ///     ArgumentException, E_ACCESSDENIED to UnauthorizedAccessException) and
+    ///     the rest to COMException. Each carries the HRESULT in Exception.HResult.
+    /// </summary>
+    private static bool IsInteropFailure(Exception ex) =>
+        ex is ExternalException
+        || ex is ArgumentException
+        || ex is UnauthorizedAccessException;
+
     private static bool IsServerGone(int hresult) =>
         hresult == unchecked((int)0x80010108) // RPC_E_DISCONNECTED
         || hresult == unchecked((int)0x800706BA) // RPC_S_SERVER_UNAVAILABLE
@@ -208,16 +270,17 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
         IMxAccessServer? mx = server;
         if (mx is not null)
         {
-            foreach (int itemHandle in itemHandles.Values)
+            foreach (AckItem item in itemLru)
             {
-                try { mx.UnAdvise(serverHandle, itemHandle); } catch { /* best effort */ }
-                try { mx.RemoveItem(serverHandle, itemHandle); } catch { /* best effort */ }
+                try { mx.UnAdvise(serverHandle, item.Handle); } catch { /* best effort */ }
+                try { mx.RemoveItem(serverHandle, item.Handle); } catch { /* best effort */ }
             }
 
             try { mx.Unregister(serverHandle); } catch { /* best effort */ }
         }
 
         itemHandles.Clear();
+        itemLru.Clear();
         server = null;
         comEventSource = null;
         object? toRelease = comObject;
@@ -228,6 +291,28 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
         }
     }
 
+    /// <summary>Caches a newly added item as most recently used, evicting the least recently used past the cap.</summary>
+    private LinkedListNode<AckItem> CacheItem(IMxAccessServer mx, string itemAddress, int itemHandle)
+    {
+        while (itemLru.Count >= maxCachedItems && itemLru.Last is { } leastRecent)
+        {
+            DropItem(mx, leastRecent);
+        }
+
+        LinkedListNode<AckItem> node = itemLru.AddFirst(new AckItem(itemAddress, itemHandle));
+        itemHandles[itemAddress] = node;
+        return node;
+    }
+
+    /// <summary>Removes a cached item and best-effort releases its MXAccess item.</summary>
+    private void DropItem(IMxAccessServer mx, LinkedListNode<AckItem> node)
+    {
+        itemHandles.Remove(node.Value.Address);
+        itemLru.Remove(node);
+        try { mx.UnAdvise(serverHandle, node.Value.Handle); } catch { /* best effort */ }
+        try { mx.RemoveItem(serverHandle, node.Value.Handle); } catch { /* best effort */ }
+    }
+
     private IMxAccessServer EnsureRegistered()
     {
         if (server is not null) return server;
@@ -276,4 +361,21 @@ public sealed class AlarmAckWriter : IAlarmAckWriter
         rows.AddRange(statusConverter.ConvertMany(pVars));
         completionCache.Record(hLMXServerHandle, phItemHandle, rows);
     }
+
+    /// <summary>A cached ack item. Plain class (not a record) for the net48 worker target.</summary>
+    private sealed class AckItem
+    {
+        public AckItem(string address, int handle)
+        {
+            Address = address;
+            Handle = handle;
+        }
+
+        public string Address { get; }
+
+        public int Handle { get; }
+
+        /// <summary>Set when the last write to this handle ended Unconfirmed; forces a fresh handle next time.</summary>
+        public bool Dirty { get; set; }
+    }
 }

src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmCommandHandler.cs

@@ -41,6 +41,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
     private readonly IMxAccessComObjectFactory? comFactory;
     private readonly Func<IReadOnlyList<AlarmSubtagTarget>, IMxAccessAlarmConsumer>? standbyFactory;
     private readonly Func<IAlarmAckWriter>? ackWriterFactory;
+    private readonly Action? ackPumpStep;
     private readonly MxAccessEventMapper mapper = new MxAccessEventMapper();
     private readonly object syncRoot = new object();
     private readonly IAlarmAckWriter deferredAckWriter;
@@ -131,13 +132,21 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
     ///     unsubscribe/resubscribe, and is disposed with the handler. Tests
     ///     inject a fake.
     /// </param>
+    /// <param name="ackPumpStep">
+    ///     Pump step the default <see cref="AlarmAckWriter"/> drives while it waits
+    ///     for OnWriteComplete. Production passes the STA runtime's
+    ///     <c>StaRuntime.PumpPendingMessages</c>, which also refreshes the STA
+    ///     watchdog heartbeat; when <see langword="null"/> a bare
+    ///     <see cref="StaMessagePump"/> is used (no heartbeat refresh).
+    /// </param>
     public AlarmCommandHandler(
         MxAccessEventQueue eventQueue,
         Func<IMxAccessAlarmConsumer> consumerFactory,
         Action? threadAffinityCheck,
         IMxAccessComObjectFactory? comFactory,
         Func<IReadOnlyList<AlarmSubtagTarget>, IMxAccessAlarmConsumer>? standbyFactory,
-        Func<IAlarmAckWriter>? ackWriterFactory = null)
+        Func<IAlarmAckWriter>? ackWriterFactory = null,
+        Action? ackPumpStep = null)
     {
         this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
         this.consumerFactory = consumerFactory ?? throw new ArgumentNullException(nameof(consumerFactory));
@@ -145,6 +154,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
         this.comFactory = comFactory;
         this.standbyFactory = standbyFactory;
         this.ackWriterFactory = ackWriterFactory;
+        this.ackPumpStep = ackPumpStep;
         this.deferredAckWriter = new DeferredAckWriter(this);
     }
 
@@ -399,12 +409,14 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
                     "Alarm acknowledgement requires an IMxAccessComObjectFactory; the alarm command handler was constructed without one.");
             }
 
-            // StaMessagePump.PumpPendingMessages is PeekMessage/DispatchMessage on
-            // the calling thread — the STA, since acks run there — which is what
-            // delivers the writer's OnWriteComplete callback.
+            // The pump step is PeekMessage/DispatchMessage on the calling thread —
+            // the STA, since acks run there — which is what delivers the writer's
+            // OnWriteComplete callback. The injected StaRuntime pump also marks STA
+            // activity so a long completion wait does not read as a hung STA.
+            Action pump = ackPumpStep ?? (() => new StaMessagePump().PumpPendingMessages());
             ackWriter = new AlarmAckWriter(
                 comFactory,
-                () => new StaMessagePump().PumpPendingMessages(),
+                pump,
                 MxAccessStaSession.ResolveWriteCompletionTimeout(),
                 AlarmAckWriter.ResolveClientName());
             return ackWriter;

src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs

@@ -48,7 +48,10 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
     private readonly IMxAccessEventSink eventSink;
     private readonly MxAccessEventQueue eventQueue;
     private readonly StaRuntime staRuntime;
-    private readonly Func<MxAccessEventQueue, Action, IMxAccessComObjectFactory, IAlarmCommandHandler>? alarmCommandHandlerFactory;
+    // Pump-aware form: the fourth argument is the STA runtime's pump step, so
+    // handler work that pumps while it waits (the alarm ack writer's
+    // OnWriteComplete wait) also refreshes the STA watchdog heartbeat.
+    private readonly Func<MxAccessEventQueue, Action, IMxAccessComObjectFactory, Action, IAlarmCommandHandler>? alarmCommandHandlerFactory;
     private StaCommandDispatcher? commandDispatcher;
     private MxAccessSession? session;
     private IAlarmCommandHandler? alarmCommandHandler;
@@ -96,6 +99,23 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
     {
     }
 
+    /// <summary>
+    /// Initializes a new instance of <see cref="MxAccessStaSession"/> with default dependencies and a
+    /// pump-aware alarm-command handler factory. Its fourth argument is a pump step bound to this
+    /// session's <see cref="StaRuntime"/> (<see cref="StaRuntime.PumpPendingMessages"/>), which
+    /// refreshes the STA watchdog heartbeat on every pump.
+    /// </summary>
+    /// <param name="pumpAwareAlarmCommandHandlerFactory">Factory that constructs the alarm-command handler.</param>
+    internal MxAccessStaSession(Func<MxAccessEventQueue, Action, IMxAccessComObjectFactory, Action, IAlarmCommandHandler> pumpAwareAlarmCommandHandlerFactory)
+    {
+        staRuntime = new StaRuntime();
+        factory = new MxAccessComObjectFactory();
+        eventQueue = new MxAccessEventQueue(MxAccessEventQueue.ResolveCapacity());
+        eventSink = new MxAccessBaseEventSink(eventQueue);
+        alarmCommandHandlerFactory = pumpAwareAlarmCommandHandlerFactory
+            ?? throw new ArgumentNullException(nameof(pumpAwareAlarmCommandHandlerFactory));
+    }
+
     /// <summary>
     /// Initializes a new instance of <see cref="MxAccessStaSession"/> with custom STA runtime and factory.
     /// </summary>
@@ -183,7 +203,9 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
         this.factory = factory ?? throw new ArgumentNullException(nameof(factory));
         this.eventSink = eventSink ?? throw new ArgumentNullException(nameof(eventSink));
         this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
-        this.alarmCommandHandlerFactory = alarmCommandHandlerFactory;
+        this.alarmCommandHandlerFactory = alarmCommandHandlerFactory is null
+            ? null
+            : (eq, affinity, comFactory, _) => alarmCommandHandlerFactory(eq, affinity, comFactory);
     }
 
     /// <summary>
@@ -288,7 +310,8 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
                     alarmCommandHandler = alarmCommandHandlerFactory(
                         eventQueue,
                         EnsureOnAlarmConsumerThread,
-                        factory);
+                        factory,
+                        () => staRuntime.PumpPendingMessages());
                 }
                 commandDispatcher = new StaCommandDispatcher(
                     staRuntime,