Merge stream E (worker: cache eviction, ReadBulk shared deadline + dedupe, alarm poll threshold, STA loop fault, unhandled-exception hook, live control-command facts)

docs/AlarmClientDiscovery.md

@@ -1051,6 +1051,18 @@ Every transition and snapshot record carries `degraded = true` and
 `source_provider = ALARM_PROVIDER_MODE_SUBTAG`. Clients that require full
 fidelity must wait for failback to the alarm manager.
 
+### Retrigger transitions
+
+`ALARM_TRANSITION_KIND_RETRIGGER` is reserved in the contract and the worker
+never emits it, in either provider mode. The alarmmgr XML reports only the
+four states `UNACK_ALM`, `ACK_ALM`, `UNACK_RTN`, and `ACK_RTN`, so a condition
+that returns to alarm after clearing (`*_RTN` → `*_ALM`) looks the same as a
+new raise, and `AlarmRecordTransitionMapper.MapTransition` reports it as
+`ALARM_TRANSITION_KIND_RAISE`. The subtag source has no re-alarm counter to
+observe either. The worker does not infer a retrigger from timing or
+from a raise that follows a clear, because the gateway forwards only
+transitions its source reports.
+
 ### Provider mode reflection
 
 When `FailoverAlarmConsumer` switches between providers, it raises

docs/MxAccessWorkerInstanceDesign.md

@@ -133,6 +133,9 @@ starts the STA, or touches MXAccess:
 | `2` | `InvalidArguments` | Required arguments are missing or unknown arguments are present. |
 | `3` | `InvalidProtocolVersion` | `--protocol-version` is not numeric or does not match the supported worker protocol. |
 | `4` | `MissingNonce` | `MXGATEWAY_WORKER_NONCE` is absent or empty. |
+| `5` | `PipeConnectionFailed` | An `IOException` or `TimeoutException` escapes the pipe client. |
+| `6` | `ProtocolViolation` | A `WorkerFrameProtocolException` escapes the pipe client. |
+| `7` | `UnhandledException` | An exception reaches the `AppDomain.UnhandledException` hook (see `docs/WorkerBootstrap.md`). |
 
 Bootstrap logs use `WorkerConsoleLogger` key/value output. `WorkerLogRedactor`
 redacts fields whose names indicate nonce, secret, password, token,
@@ -270,12 +273,33 @@ is still responsive. Shutdown marks the runtime as closing, wakes the pump,
 rejects new commands, cancels queued work, uninitializes COM on the STA, and
 waits for the thread to exit.
 
+`ProcessQueuedCommands` pumps pending Windows messages between queued work
+items whenever more work remains, not only after the whole queue drains. A
+burst of queued commands would otherwise hold every inbound COM callback
+(`OnDataChange`, `OnWriteComplete`) behind the entire backlog.
+
+If the STA loop itself throws after `Start` has returned (a failed
+`MsgWaitForMultipleObjectsEx`, or an exception escaping a dispatched message),
+`StaRuntime` records the exception as `LoopFault`, fails every later
+`InvokeAsync` with an `InvalidOperationException` wrapping it (never the
+graceful `StaRuntimeShutdownException`), cancels queued work, and invokes the
+handler registered through `RegisterLoopFaultHandler` — unless a requested
+shutdown is already in progress, in which case the death is part of a
+deliberate teardown and is not reported. `MxAccessStaSession` registers a
+handler that records a `StaHung` `WorkerFault` on the event queue (diagnostic
+text carries the exception type and HRESULT only, never the message),
+so the IPC drain loop sends the fault to the gateway and ends the pipe session
+and the worker process. Before this, a post-start loop death was stored in the
+startup-exception field that only `Start` reads, leaving a live worker whose
+commands queued forever. Startup failures still surface from `Start` itself.
+
 ### Inner Completion Waits
 
 Two commands hold the STA while waiting for a COM event they just provoked: the
 unary write path waits for its `OnWriteComplete`
-(`MxAccessWriteCompletionCache.TryWaitForCompletion`), and `ReadBulk` waits per
-tag for the first `OnDataChange` (`MxAccessValueCache.TryWaitForUpdate`). Both
+(`MxAccessWriteCompletionCache.TryWaitForCompletion`), and `ReadBulk` waits for
+the first `OnDataChange` of every snapshot tag
+(`MxAccessValueCache.TryWaitForUpdates`). Both
 run the same loop shape as the outer pump, for the same reason — the event they
 are waiting for *is* a Windows message, so the thread must keep dispatching to
 receive it:
@@ -297,7 +321,7 @@ The idle slice is a Win32 wait (`StaWaitHelper.WaitForSignalOrMessages`), never
 `Thread.Sleep`. A sleeping STA pumps no messages, so a sleep-polled loop could
 only dispatch the awaited COM event at poll-tick granularity while stalling
 *every other* event for the same tick — up to 1.5 s for a write completion and
-up to `timeout_ms` per tag for `ReadBulk`. The Win32 wait returns the instant a
+up to `timeout_ms` for `ReadBulk`. The Win32 wait returns the instant a
 message needs pumping, so the apartment dispatches continuously for the whole
 wait. Each cache also sets an `AutoResetEvent` from its update path (outside the
 cache lock) so a cross-thread producer wakes the waiter immediately; in the live
@@ -317,8 +341,29 @@ nothing wakes the wait — a process with no STA message queue (unit tests drive
 these caches from ordinary threads, standing in for the STA by updating the
 cache from a fake `pumpStep`) must not block for a full poll interval. Timeouts,
 deadline math, and return values are unchanged by the wait mechanism: an expired
-write wait still yields the empty-`statuses` unconfirmed reply, and an expired
-per-tag `ReadBulk` wait still reports its own timeout.
+write wait still yields the empty-`statuses` unconfirmed reply, and every
+`ReadBulk` tag that did not deliver before the deadline still reports its own
+timeout.
+
+`ReadBulk` runs its snapshot lifecycle in phases so the STA is held for at most
+one `timeout_ms` per command, not one per tag: (1) every tag is validated and
+probed against the cached-and-advised fast path; (2) every remaining tag is
+added and advised; (3) one wait with a shared deadline collects the first
+`OnDataChange` per snapshot handle and returns as soon as all have delivered;
+(4) every snapshot handle is unadvised and removed, best-effort. Per-tag
+results are unchanged: request order, the blank-address failure, the cached
+result, an `AddItem` failure (item handle 0) or `Advise` failure (its own item
+handle) failing only that tag, and the `ReadBulk timed out after <n> ms waiting
+for first OnDataChange.` message for each tag that never delivered. The cache
+probes all run before any snapshot `AddItem`, so a later tag in the same
+request can never be served from an earlier tag's snapshot subscription.
+When MXAccess returns the same item handle for two tags in one batch (a
+duplicate address), the handle is advised, unadvised, and removed once, and
+every duplicate receives the shared outcome. Phases 2 and 4 make one or two
+COM calls per tag without waiting, so they call the pump step every
+`ReadBulkPumpInterval` (16) tags; a large batch therefore keeps dispatching
+inbound callbacks and refreshing `LastStaActivityUtc` instead of drifting
+toward `HeartbeatStuckCeiling`.
 
 The write wait's budget is `MxGateway:Worker:WriteCompletionWaitMilliseconds`
 (default 1500). It applies to `Write`, `Write2`, `WriteSecured`, and
@@ -671,7 +716,20 @@ The worker records plain and supervisory advice separately only after the COM
 call returns normally. Successful `UnAdvise` removes all tracked advice for the
 server and item pair because the public MXAccess cleanup method has no plain
 versus supervisory selector. Successful `RemoveItem` and `Unregister` also clear
-related advice state from the worker registry. Failed advice and cleanup calls
+related advice state from the worker registry, and evict the per-handle
+caches the event sink fills: `RemoveItem` drops the handle pair's
+`MxAccessValueCache` and `MxAccessWriteCompletionCache` entries, and
+`Unregister` drops every entry recorded under the server handle (MXAccess
+removes the server's items with it, so the per-item eviction never runs for
+them). Without this, an `AddItem` → `Write` → `RemoveItem` churn grew the
+completion cache by one entry per cycle for the life of the worker.
+`AddItem`, `AddItem2`, and `AddBufferedItem` also drop both cache entries for
+the handle they receive when the registry does not already track it. An
+`OnDataChange` or `OnWriteComplete` that was queued when `RemoveItem` or
+`Unregister` ran is dispatched after the eviction and recreates the entry;
+when MXAccess later reuses that item id, `ReadBulk`'s cached path would
+otherwise serve the stale value as `was_cached = true`. A handle the
+registry still tracks belongs to a live item, so its entries are kept. Failed advice and cleanup calls
 leave registry state unchanged so diagnostics continue to reflect the last
 successful MXAccess-owned state transition.
 
@@ -905,6 +963,23 @@ the STA more than 75 s without pumping *should* fault — that is the ceiling's
 contract). The flag is named generically so any future non-dispatcher STA work
 reuses it.
 
+A `PollOnce` that throws does not fault the session on the first failure. The
+poll loop counts consecutive failures and records the alarm-poll fault
+(`MxaccessEventConversionFailed`) only when the count reaches
+`MxAccessStaSession.DefaultAlarmPollFailureThreshold` (3, the same default as
+`AlarmFailoverConfig.consecutive_failure_threshold`); a successful poll resets
+the count. Each failure below the threshold writes one
+`level=Warning event=AlarmPollFailed` line to stderr carrying the exception
+type, HRESULT, and counts only — the exception message is left out because it
+can echo alarm or tag names. A single transient COM failure in
+`GetXmlCurrentAlarms2` used to fault the whole session, taking every data
+subscription with it; a persistent failure, including an STA-affinity
+regression that fails every tick, still faults within three poll intervals.
+A poll failure seen after the STA loop itself has died (`StaRuntime.LoopFault`
+is set) ends the poll loop without recording anything: the loop-fault handler
+has already recorded the `StaHung` fault, and the failure is only the
+fast-fail `InvokeAsync` returns afterwards.
+
 The in-flight-command suppression itself is bounded by
 `WorkerPipeSessionOptions.HeartbeatStuckCeiling` (default 75 seconds = 5 ×
 `HeartbeatGrace`). The motivating case for the suppression is a legitimately

docs/WorkerBootstrap.md

@@ -117,8 +117,13 @@ public static WorkerBootstrapResult Failure(WorkerExitCode exitCode, IEnumerable
 | `MissingNonce` | 4 | The `MXGATEWAY_WORKER_NONCE` environment variable is null, empty, or whitespace. |
 | `PipeConnectionFailed` | 5 | An `IOException` or `TimeoutException` escapes the pipe client. |
 | `ProtocolViolation` | 6 | A `WorkerFrameProtocolException` escapes the pipe client. |
+| `UnhandledException` | 7 | An exception reaches `AppDomain.CurrentDomain.UnhandledException` — thrown on a thread outside `WorkerApplication.Run`'s `try/catch` (a background task continuation, a timer, a COM callback thread). |
 
-`InvalidArguments`, `InvalidProtocolVersion`, and `MissingNonce` originate in the parser; the others originate in `WorkerApplication.Run`'s `try/catch` around the pipe client.
+`InvalidArguments`, `InvalidProtocolVersion`, and `MissingNonce` originate in the parser; `UnhandledException` originates in the process-wide hook described below; the others originate in `WorkerApplication.Run`'s `try/catch` around the pipe client.
+
+### Unhandled-exception hook
+
+The production entry point, `WorkerApplication.Run(string[])`, subscribes `AppDomain.CurrentDomain.UnhandledException` before it parses arguments. The handler (`WorkerApplication.HandleUnhandledException`) logs one `WorkerUnhandledException` error line and calls `Environment.Exit(7)`. Without it, an exception on a thread that `Run` does not own ended the process with the CLR's crash code and no worker log line, so the gateway saw an unexplained exit. The line carries `exit_code`, `exception_type`, and `is_terminating` only: the exception message and stack are left out because they can carry tag values or credentials passed through a command, the same rule every other worker failure line follows. A failure inside the logger is swallowed so the exit always happens. The injectable `Run` overloads that tests call do not install the hook, because it is process-global.
 
 ## Environment Abstraction
 

docs/WorkerSta.md

@@ -86,10 +86,16 @@ StaWorkItem<T> workItem = new(command, cancellationToken);
 
 lock (gate)
 {
+    if (loopFault is not null)
+    {
+        return Task.FromException<T>(new InvalidOperationException(
+            "The worker STA loop terminated unexpectedly.",
+            loopFault));
+    }
+
     if (shutdownRequested)
     {
-        return Task.FromException<T>(
-            new InvalidOperationException("The worker STA runtime is shutting down."));
+        return Task.FromException<T>(new StaRuntimeShutdownException());
     }
 
     commandQueue.Enqueue(workItem);
@@ -101,7 +107,11 @@ return workItem.Task;
 
 `StaWorkItem<T>` uses an `Interlocked.CompareExchange` on `started` so that exactly one of three outcomes happens: the STA thread executes the delegate, an external `CancellationToken` cancellation fires first, or `CancelBeforeExecution` runs during shutdown. `Execute` runs on the STA thread, sets `Completion` from `command()`, and propagates exceptions through `TrySetException` so the awaiting caller observes them.
 
-`ProcessQueuedCommands` is the only consumer of the queue and runs on the STA thread, so each work item is guaranteed to execute on the apartment that owns the COM object.
+`ProcessQueuedCommands` is the only consumer of the queue and runs on the STA thread, so each work item is guaranteed to execute on the apartment that owns the COM object. It calls `PumpPendingMessages` between work items whenever more work is queued, so a backlog of commands cannot hold inbound COM callbacks until the whole queue drains.
+
+### Loop failure after start
+
+An exception thrown by the loop after `Start()` returned — `WaitForWorkOrMessages` reporting `WAIT_FAILED`, or an exception escaping a dispatched message — is not a startup failure, and nothing reads `startupException` after `Start()`. `ThreadMain` instead stores it as `LoopFault` under `gate`, so every later `InvokeAsync` fails immediately with an `InvalidOperationException` whose inner exception is the loop fault, and work already queued is cancelled by the `finally` block. It then invokes the handler registered with `RegisterLoopFaultHandler`; a handler registered after the loop died runs immediately on the registering thread. Neither happens when `Shutdown` was already requested, because a loop that throws during a deliberate teardown is not a worker fault. `MxAccessStaSession` registers a handler that records a `StaHung` `WorkerFault` on the event queue, which the IPC drain loop sends to the gateway before ending the pipe session and the worker process.
 
 ## Command Dispatch
 

gateway.md

@@ -571,8 +571,15 @@ returns one result per requested entry — per-entry failures populate
   advised (`was_cached = true`, no subscription side-effects), or by
   taking a full `AddItem` + `Advise` + wait-for-first-OnDataChange +
   `UnAdvise` + `RemoveItem` snapshot lifecycle when no live subscription
-  exists (`was_cached = false`). Per-tag timeouts surface as
-  `was_successful = false` rather than throwing. The cache lives on the
+  exists (`was_cached = false`). `timeout_ms` is one shared deadline for
+  the whole command, not a per-tag budget: the worker adds and advises
+  every uncached tag first, waits once for all of them, then tears them
+  all down, so the STA is held for at most one `timeout_ms` (plus the
+  add/advise and teardown COM calls, during which it pumps every 16 tags).
+  Each tag that has not delivered by the deadline surfaces its own
+  `was_successful = false` timeout rather than throwing. Duplicate
+  addresses that MXAccess resolves to the same item handle share one
+  subscription and receive the same result. The cache lives on the
   worker's `MxAccessValueCache`, populated by `MxAccessBaseEventSink` on
   every `OnDataChange` after the event clears the outbound queue.
   Required scope: `invoke:read`. `timeout_ms == 0` uses the worker's

src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs

@@ -1052,6 +1052,231 @@ public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output)
         }
     }
 
+    /// <summary>
+    /// TST-05: the real worker answers the four non-terminating control commands off-STA in
+    /// <c>WorkerPipeSession.HandleControlCommandAsync</c>. The default suite only ever sees the
+    /// FakeWorkerHarness canned replies, so this drives <c>Ping</c>, <c>GetSessionState</c>,
+    /// <c>GetWorkerInfo</c>, and <c>DrainEvents</c> through the gateway against a live worker and
+    /// asserts each reply carries real worker state rather than an <c>INVALID_REQUEST</c>.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [LiveMxAccessFact]
+    public async Task GatewaySession_WithLiveWorker_ControlCommands_ReturnRealWorkerState()
+    {
+        string workerExecutablePath = IntegrationTestEnvironment.ResolveLiveMxAccessWorkerExecutablePath();
+        Assert.True(
+            File.Exists(workerExecutablePath),
+            $"Live MXAccess worker executable was not found at {workerExecutablePath}. Build the worker or set {IntegrationTestEnvironment.LiveMxAccessWorkerExecutableVariableName}.");
+
+        TestWorkerProcessFactory processFactory = new(output);
+        await using GatewayServiceFixture fixture = new(workerExecutablePath, processFactory, output);
+
+        string? sessionId = null;
+
+        try
+        {
+            OpenSessionReply openReply = await fixture.Service.OpenSession(
+                new OpenSessionRequest
+                {
+                    ClientSessionName = "live-mxaccess-control",
+                    ClientCorrelationId = "live-open-control",
+                    CommandTimeout = Duration.FromTimeSpan(CommandTimeout),
+                },
+                new TestServerCallContext()).ConfigureAwait(false);
+
+            sessionId = openReply.SessionId;
+            Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code);
+            Assert.True(openReply.WorkerProcessId > 0);
+
+            MxCommandReply pingReply = await fixture.Service.Invoke(
+                CreateControlRequest(sessionId, "live-ping", new MxCommand
+                {
+                    Kind = MxCommandKind.Ping,
+                    Ping = new PingCommand { Message = "live-ping-echo" },
+                }),
+                new TestServerCallContext()).ConfigureAwait(false);
+            LogReply("Ping", pingReply);
+            Assert.Equal(ProtocolStatusCode.Ok, pingReply.ProtocolStatus.Code);
+            Assert.Equal(MxCommandKind.Ping, pingReply.Kind);
+
+            MxCommandReply stateReply = await fixture.Service.Invoke(
+                CreateControlRequest(sessionId, "live-get-session-state", new MxCommand
+                {
+                    Kind = MxCommandKind.GetSessionState,
+                    GetSessionState = new GetSessionStateCommand(),
+                }),
+                new TestServerCallContext()).ConfigureAwait(false);
+            LogReply("GetSessionState", stateReply);
+            Assert.Equal(ProtocolStatusCode.Ok, stateReply.ProtocolStatus.Code);
+            Assert.Equal(MxCommandKind.GetSessionState, stateReply.Kind);
+            Assert.NotNull(stateReply.SessionState);
+            Assert.Equal(SessionState.Ready, stateReply.SessionState.State);
+
+            MxCommandReply infoReply = await fixture.Service.Invoke(
+                CreateControlRequest(sessionId, "live-get-worker-info", new MxCommand
+                {
+                    Kind = MxCommandKind.GetWorkerInfo,
+                    GetWorkerInfo = new GetWorkerInfoCommand(),
+                }),
+                new TestServerCallContext()).ConfigureAwait(false);
+            LogReply("GetWorkerInfo", infoReply);
+            Assert.Equal(ProtocolStatusCode.Ok, infoReply.ProtocolStatus.Code);
+            Assert.Equal(MxCommandKind.GetWorkerInfo, infoReply.Kind);
+            Assert.NotNull(infoReply.WorkerInfo);
+            output.WriteLine(
+                $"GetWorkerInfo pid={infoReply.WorkerInfo.WorkerProcessId} version={infoReply.WorkerInfo.WorkerVersion} "
+                + $"progid={infoReply.WorkerInfo.MxaccessProgid} clsid={infoReply.WorkerInfo.MxaccessClsid}");
+            Assert.Equal(openReply.WorkerProcessId, infoReply.WorkerInfo.WorkerProcessId);
+            Assert.False(string.IsNullOrWhiteSpace(infoReply.WorkerInfo.WorkerVersion));
+            Assert.False(string.IsNullOrWhiteSpace(infoReply.WorkerInfo.MxaccessProgid));
+            Assert.False(string.IsNullOrWhiteSpace(infoReply.WorkerInfo.MxaccessClsid));
+
+            // No StreamEvents subscriber is attached, so whatever the worker queued since start
+            // is still drainable; the reply must carry the DrainEvents payload (possibly empty).
+            MxCommandReply drainReply = await fixture.Service.Invoke(
+                CreateControlRequest(sessionId, "live-drain-events", new MxCommand
+                {
+                    Kind = MxCommandKind.DrainEvents,
+                    DrainEvents = new DrainEventsCommand { MaxEvents = 16 },
+                }),
+                new TestServerCallContext()).ConfigureAwait(false);
+            LogReply("DrainEvents", drainReply);
+            Assert.Equal(ProtocolStatusCode.Ok, drainReply.ProtocolStatus.Code);
+            Assert.Equal(MxCommandKind.DrainEvents, drainReply.Kind);
+            Assert.Equal(MxCommandReply.PayloadOneofCase.DrainEvents, drainReply.PayloadCase);
+            Assert.True(drainReply.DrainEvents.Events.Count <= 16);
+            output.WriteLine($"DrainEvents drained={drainReply.DrainEvents.Events.Count}");
+
+            // A control command must not disturb the session: it is still Ready afterwards.
+            Assert.True(fixture.TryGetSession(sessionId, out GatewaySession? session));
+            Assert.Equal(SessionState.Ready, session!.State);
+        }
+        finally
+        {
+            await ShutDownAsync(fixture, processFactory, sessionId, streamTask: null).ConfigureAwait(false);
+        }
+    }
+
+    /// <summary>
+    /// TST-05: <c>ShutdownWorker</c> writes its OK reply before the worker shuts down, so the
+    /// reply reaches the caller (a reply written after the pipe closed would surface as a
+    /// pipe-disconnected failure instead), the worker process then exits on its own, and a later
+    /// command on the session does not succeed. Uses its own fixture and worker because it
+    /// terminates the worker.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [LiveMxAccessFact]
+    public async Task GatewaySession_WithLiveWorker_ShutdownWorker_RepliesOkThenWorkerExits()
+    {
+        string workerExecutablePath = IntegrationTestEnvironment.ResolveLiveMxAccessWorkerExecutablePath();
+        Assert.True(
+            File.Exists(workerExecutablePath),
+            $"Live MXAccess worker executable was not found at {workerExecutablePath}. Build the worker or set {IntegrationTestEnvironment.LiveMxAccessWorkerExecutableVariableName}.");
+
+        TestWorkerProcessFactory processFactory = new(output);
+        await using GatewayServiceFixture fixture = new(workerExecutablePath, processFactory, output);
+        using RecordingServerStreamWriter<MxEvent> eventWriter = new();
+
+        string? sessionId = null;
+        Task? streamTask = null;
+        using CancellationTokenSource streamCancellation = new();
+
+        try
+        {
+            OpenSessionReply openReply = await fixture.Service.OpenSession(
+                new OpenSessionRequest
+                {
+                    ClientSessionName = "live-mxaccess-shutdown-worker",
+                    ClientCorrelationId = "live-open-shutdown-worker",
+                    CommandTimeout = Duration.FromTimeSpan(CommandTimeout),
+                },
+                new TestServerCallContext()).ConfigureAwait(false);
+
+            sessionId = openReply.SessionId;
+            Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code);
+
+            // The gateway observes a worker's pipe closing through the event-stream read loop
+            // (the same path the abnormal-exit fact relies on); without a subscriber the session
+            // would sit in Ready until the heartbeat grace expired.
+            streamTask = fixture.Service.StreamEvents(
+                new StreamEventsRequest { SessionId = sessionId },
+                eventWriter,
+                new TestServerCallContext(streamCancellation.Token));
+
+            MxCommandReply shutdownReply = await fixture.Service.Invoke(
+                CreateControlRequest(sessionId, "live-shutdown-worker", new MxCommand
+                {
+                    Kind = MxCommandKind.ShutdownWorker,
+                    ShutdownWorker = new ShutdownWorkerCommand
+                    {
+                        GracePeriod = Duration.FromTimeSpan(TimeSpan.FromSeconds(5)),
+                    },
+                }),
+                new TestServerCallContext()).ConfigureAwait(false);
+            LogReply("ShutdownWorker", shutdownReply);
+            Assert.Equal(ProtocolStatusCode.Ok, shutdownReply.ProtocolStatus.Code);
+            Assert.Equal(MxCommandKind.ShutdownWorker, shutdownReply.Kind);
+
+            // The worker exits on its own — no CloseSession, no kill. WaitForProcessesAsync
+            // throws on timeout, which fails the test.
+            await processFactory.WaitForProcessesAsync(StreamShutdownTimeout).ConfigureAwait(false);
+
+            // The session can no longer serve commands once its worker is gone: either the
+            // gateway rejects the call outright or the reply is not Ok.
+            MxCommandReply? afterReply = null;
+            Exception? afterException = null;
+            try
+            {
+                afterReply = await fixture.Service.Invoke(
+                    CreateControlRequest(sessionId, "live-ping-after-shutdown", new MxCommand
+                    {
+                        Kind = MxCommandKind.Ping,
+                        Ping = new PingCommand(),
+                    }),
+                    new TestServerCallContext()).ConfigureAwait(false);
+                LogReply("PingAfterShutdown", afterReply);
+            }
+            catch (Exception exception)
+            {
+                afterException = exception;
+                output.WriteLine($"PingAfterShutdown rejected: {exception.GetType().Name}: {exception.Message}");
+            }
+
+            Assert.True(
+                afterException is not null || afterReply!.ProtocolStatus.Code != ProtocolStatusCode.Ok,
+                "A command sent after ShutdownWorker must not succeed against the exited worker.");
+
+            // Logged, not asserted: the gateway's WorkerClient moves to Closed on the shutdown
+            // ack, but GatewaySession.State is not driven from that transition today, so the
+            // public session state still reads Ready here (observed live on windev). That is a
+            // gateway-side gap, not a worker one — this fact pins the worker half of the
+            // contract: OK reply first, then a clean exit.
+            string observedState = fixture.TryGetSession(sessionId, out GatewaySession? session)
+                ? session.State.ToString()
+                : "<removed>";
+            output.WriteLine($"ShutdownWorker observed_session_state={observedState}");
+        }
+        finally
+        {
+            // The worker is gone; a CloseSession round-trip would only log a cleanup failure.
+            streamCancellation.Cancel();
+            await ShutDownAsync(fixture, processFactory, sessionId: null, streamTask).ConfigureAwait(false);
+        }
+    }
+
+    private static MxCommandRequest CreateControlRequest(
+        string sessionId,
+        string correlationId,
+        MxCommand command)
+    {
+        return new MxCommandRequest
+        {
+            SessionId = sessionId,
+            ClientCorrelationId = correlationId,
+            Command = command,
+        };
+    }
+
     /// <summary>
     /// Closes the session and drains the event stream / worker processes without letting a
     /// cleanup timeout mask the original failure from the test body.

src/ZB.MOM.WW.MxGateway.Worker.Tests/Bootstrap/WorkerApplicationTests.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using ZB.MOM.WW.MxGateway.Contracts;
@@ -121,6 +122,49 @@ public sealed class WorkerApplicationTests
         Assert.Equal(typeof(InvalidOperationException).FullName, entry.Fields["exception_type"]);
     }
 
+    /// <summary>
+    ///     WRK-09: the AppDomain unhandled-exception hook logs the exception type
+    ///     (never its message) and exits with the dedicated exit code.
+    /// </summary>
+    [Fact]
+    public void HandleUnhandledException_LogsTypeOnlyAndExitsWithUnhandledExceptionCode()
+    {
+        MemoryWorkerLogger logger = new();
+        int? exitCode = null;
+
+        ZB.MOM.WW.MxGateway.Worker.WorkerApplication.HandleUnhandledException(
+            logger,
+            new InvalidOperationException("password=hunter2"),
+            isTerminating: true,
+            code => exitCode = code);
+
+        Assert.Equal((int)WorkerExitCode.UnhandledException, exitCode);
+        Assert.Equal(7, (int)WorkerExitCode.UnhandledException);
+        MemoryWorkerLogEntry entry = Assert.Single(logger.Entries);
+        Assert.Equal("Error", entry.Level);
+        Assert.Equal("WorkerUnhandledException", entry.EventName);
+        Assert.Equal(typeof(InvalidOperationException).FullName, entry.Fields["exception_type"]);
+        Assert.True(entry.Fields["is_terminating"] is true);
+        Assert.DoesNotContain(
+            entry.Fields.Values,
+            value => value?.ToString()?.Contains("hunter2") == true);
+    }
+
+    /// <summary>A failing logger must not prevent the defined exit.</summary>
+    [Fact]
+    public void HandleUnhandledException_WhenLoggerThrows_StillExits()
+    {
+        int? exitCode = null;
+
+        ZB.MOM.WW.MxGateway.Worker.WorkerApplication.HandleUnhandledException(
+            new ThrowingLogger(),
+            exceptionObject: null,
+            isTerminating: false,
+            code => exitCode = code);
+
+        Assert.Equal((int)WorkerExitCode.UnhandledException, exitCode);
+    }
+
     private static string[] ValidArgs(string? protocolVersion = null)
     {
         return
@@ -171,4 +215,15 @@ public sealed class WorkerApplicationTests
             throw _exception;
         }
     }
+
+    private sealed class ThrowingLogger : IWorkerLogger
+    {
+        /// <inheritdoc />
+        public void Information(string eventName, IReadOnlyDictionary<string, object?> fields) =>
+            throw new InvalidOperationException("logger failed");
+
+        /// <inheritdoc />
+        public void Error(string eventName, IReadOnlyDictionary<string, object?> fields) =>
+            throw new InvalidOperationException("logger failed");
+    }
 }

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

@@ -499,6 +499,40 @@ public sealed class AlarmCommandHandlerTests
         Assert.Equal(AlarmProviderMode.Subtag, evt.OnAlarmProviderModeChanged.Mode);
     }
 
+    /// <summary>
+    ///     In auto mode the primary consumer is built before the standby. When
+    ///     building the standby throws, the primary is owned by nothing — the
+    ///     failed Subscribe never reaches the dispatcher that would dispose it —
+    ///     so the handler must dispose it rather than leak its COM object, and
+    ///     the original failure must still propagate.
+    /// </summary>
+    [Fact]
+    public void Subscribe_AutoModeWhenStandbyBuildThrows_DisposesPrimaryAndRethrows()
+    {
+        FakeConsumer primary = new FakeConsumer();
+        InvalidOperationException standbyFailure = new InvalidOperationException("standby build failed");
+        AlarmCommandHandler handler = new AlarmCommandHandler(
+            new MxAccessEventQueue(),
+            () => primary,
+            threadAffinityCheck: null,
+            comFactory: null,
+            standbyFactory: _ => throw standbyFailure);
+
+        SubscribeAlarmsCommand command = new SubscribeAlarmsCommand
+        {
+            SubscriptionExpression = @"\\HOST\Galaxy!Area",
+            ForcedMode = AlarmProviderMode.Unspecified,
+        };
+        command.WatchList.Add(NewTarget("Galaxy!Area.Tank01.Level.HiHi"));
+
+        InvalidOperationException thrown = Assert.Throws<InvalidOperationException>(
+            () => handler.Subscribe(command, "s1"));
+
+        Assert.Same(standbyFailure, thrown);
+        Assert.True(primary.Disposed, "The primary consumer built before the failing standby must be disposed.");
+        Assert.False(handler.IsSubscribed);
+    }
+
     /// <summary>An explicitly fallback-capable auto subscription builds a replaceable standby even when discovery is initially empty.</summary>
     [Fact]
     public void Subscribe_AutoModeWithEmptyWatchListAndFallbackEnabled_BuildsReplaceableStandby()

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

@@ -102,6 +102,22 @@ public sealed class AlarmRecordTransitionMapperTests
         Assert.Equal(expected, AlarmRecordTransitionMapper.MapTransition(previous, current));
     }
 
+    /// <summary>
+    ///     RETRIGGER is reserved and never synthesized: no state pair — including
+    ///     cleared-to-active, the re-alarm shape — maps to it.
+    /// </summary>
+    [Fact]
+    public void MapTransition_ForEveryStatePair_NeverProducesRetrigger()
+    {
+        foreach (MxAlarmStateKind previous in System.Enum.GetValues(typeof(MxAlarmStateKind)))
+        {
+            foreach (MxAlarmStateKind current in System.Enum.GetValues(typeof(MxAlarmStateKind)))
+            {
+                Assert.NotEqual(AlarmTransitionKind.Retrigger, AlarmRecordTransitionMapper.MapTransition(previous, current));
+            }
+        }
+    }
+
     /// <summary>Verifies that valid XML fields assemble into correct UTC timestamp.</summary>
     [Fact]
     public void ParseTransitionTimestampUtc_WithValidXmlFields_AssemblesUtc()

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

@@ -0,0 +1,591 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using Google.Protobuf.Collections;
+using Google.Protobuf.WellKnownTypes;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Worker.MxAccess;
+using ZB.MOM.WW.MxGateway.Worker.Tests.TestSupport;
+
+namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess;
+
+/// <summary>
+///     Direct tests of <see cref="MxAccessSession"/> handle-lifetime cache
+///     eviction and the phased ReadBulk snapshot path, driven through
+///     <see cref="MxAccessSession.CreateForTesting"/> with a recording fake
+///     server. Nothing here needs an STA: the session methods are plain calls,
+///     and ReadBulk's pump step is a test delegate that stands in for the COM
+///     OnDataChange dispatch by writing the value cache directly.
+/// </summary>
+public sealed class MxAccessSessionTests
+{
+    private const string ReadBulkTimeoutText = "timed out";
+
+    /// <summary>
+    ///     RemoveItem must evict the handle pair's OnWriteComplete completion
+    ///     entry as well as its last-value entry; otherwise an
+    ///     AddItem -> Write -> RemoveItem churn grows the completion cache by one
+    ///     entry per cycle for the life of the worker.
+    /// </summary>
+    [Fact]
+    public void RemoveItem_EvictsValueAndWriteCompletionEntriesForTheHandle()
+    {
+        RecordingMxAccessServer server = new();
+        CompletionCacheSink sink = new();
+        MxAccessValueCache valueCache = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, sink, valueCache: valueCache);
+
+        int serverHandle = session.Register("client");
+        int keep = session.AddItem(serverHandle, "Tag.Keep");
+        int drop = session.AddItem(serverHandle, "Tag.Drop");
+        RecordCompletion(sink.WriteCompletionCache, serverHandle, keep);
+        RecordCompletion(sink.WriteCompletionCache, serverHandle, drop);
+        valueCache.Set(serverHandle, keep, BuildDataChange(serverHandle, keep, 1));
+        valueCache.Set(serverHandle, drop, BuildDataChange(serverHandle, drop, 2));
+
+        session.RemoveItem(serverHandle, drop);
+
+        Assert.Equal(1, sink.WriteCompletionCache.Count);
+        Assert.Equal(0UL, sink.WriteCompletionCache.CurrentVersion(serverHandle, drop));
+        Assert.Equal(1UL, sink.WriteCompletionCache.CurrentVersion(serverHandle, keep));
+        Assert.False(valueCache.TryGet(serverHandle, drop, out _));
+        Assert.True(valueCache.TryGet(serverHandle, keep, out _));
+    }
+
+    /// <summary>
+    ///     A long-lived writer that keeps adding, writing, and removing items
+    ///     must leave the completion cache empty — the leak this fix closes.
+    /// </summary>
+    [Fact]
+    public void AddItemWriteRemoveItemChurn_DoesNotGrowWriteCompletionCache()
+    {
+        RecordingMxAccessServer server = new();
+        CompletionCacheSink sink = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, sink);
+        int serverHandle = session.Register("client");
+
+        for (int cycle = 0; cycle < 50; cycle++)
+        {
+            int itemHandle = session.AddItem(serverHandle, "Tag.Churn");
+            session.Write(serverHandle, itemHandle, cycle, userId: 0);
+
+            // Stand in for the sink's OnWriteComplete post-publish hook.
+            RecordCompletion(sink.WriteCompletionCache, serverHandle, itemHandle);
+            session.RemoveItem(serverHandle, itemHandle);
+        }
+
+        Assert.Equal(0, sink.WriteCompletionCache.Count);
+    }
+
+    /// <summary>
+    ///     Unregister drops every item under the server in one MXAccess call,
+    ///     so the per-item eviction never runs for them: both caches must be
+    ///     evicted by server handle, and other servers' entries kept.
+    /// </summary>
+    [Fact]
+    public void Unregister_EvictsValueAndWriteCompletionEntriesForThatServerOnly()
+    {
+        RecordingMxAccessServer server = new();
+        CompletionCacheSink sink = new();
+        MxAccessValueCache valueCache = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, sink, valueCache: valueCache);
+
+        int doomed = session.Register("doomed");
+        int survivor = session.Register("survivor");
+        int doomedItemA = session.AddItem(doomed, "Tag.A");
+        int doomedItemB = session.AddItem(doomed, "Tag.B");
+        int survivorItem = session.AddItem(survivor, "Tag.C");
+        RecordCompletion(sink.WriteCompletionCache, doomed, doomedItemA);
+        RecordCompletion(sink.WriteCompletionCache, doomed, doomedItemB);
+        RecordCompletion(sink.WriteCompletionCache, survivor, survivorItem);
+        valueCache.Set(doomed, doomedItemA, BuildDataChange(doomed, doomedItemA, 1));
+        valueCache.Set(doomed, doomedItemB, BuildDataChange(doomed, doomedItemB, 2));
+        valueCache.Set(survivor, survivorItem, BuildDataChange(survivor, survivorItem, 3));
+
+        session.Unregister(doomed);
+
+        Assert.Equal(1, sink.WriteCompletionCache.Count);
+        Assert.Equal(1UL, sink.WriteCompletionCache.CurrentVersion(survivor, survivorItem));
+        Assert.Equal(1, valueCache.Count);
+        Assert.True(valueCache.TryGet(survivor, survivorItem, out _));
+        Assert.False(valueCache.TryGet(doomed, doomedItemA, out _));
+        Assert.False(valueCache.TryGet(doomed, doomedItemB, out _));
+    }
+
+    /// <summary>
+    ///     N uncached tags that never deliver an OnDataChange must cost one
+    ///     shared timeout, not N x timeout, while each still reports its own
+    ///     timed-out failure with its own item handle.
+    /// </summary>
+    [Fact]
+    public void ReadBulk_UncachedTagsThatNeverDeliver_WaitOneSharedTimeoutAndEachTimesOut()
+    {
+        RecordingMxAccessServer server = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink());
+        int serverHandle = session.Register("client");
+        string[] tags = { "Tag.1", "Tag.2", "Tag.3", "Tag.4", "Tag.5" };
+        TimeSpan timeout = TimeSpan.FromMilliseconds(300);
+
+        Stopwatch stopwatch = Stopwatch.StartNew();
+        IReadOnlyList<BulkReadResult> results = session.ReadBulk(serverHandle, tags, timeout, () => { });
+        stopwatch.Stop();
+
+        // Serial would be >= 5 x 300 ms = 1500 ms. Allow generous slack for a
+        // loaded CI host while still failing the serial shape.
+        Assert.True(
+            stopwatch.Elapsed < TimeSpan.FromMilliseconds(1200),
+            $"ReadBulk took {stopwatch.Elapsed.TotalMilliseconds:F0} ms; expected one shared {timeout.TotalMilliseconds:F0} ms wait.");
+        Assert.True(stopwatch.Elapsed >= TimeSpan.FromMilliseconds(250));
+
+        Assert.Equal(tags.Length, results.Count);
+        HashSet<int> handles = new();
+        for (int index = 0; index < tags.Length; index++)
+        {
+            BulkReadResult result = results[index];
+            Assert.Equal(tags[index], result.TagAddress);
+            Assert.False(result.WasSuccessful);
+            Assert.False(result.WasCached);
+            Assert.Equal(serverHandle, result.ServerHandle);
+            Assert.Equal(server.HandleForTag(tags[index]), result.ItemHandle);
+            Assert.Equal(
+                $"ReadBulk timed out after {timeout.TotalMilliseconds:F0} ms waiting for first OnDataChange.",
+                result.ErrorMessage);
+            Assert.True(handles.Add(result.ItemHandle));
+        }
+
+        // Every snapshot handle is still torn down.
+        foreach (int handle in handles)
+        {
+            Assert.Contains($"UnAdvise:{serverHandle}:{handle}", server.Operations);
+            Assert.Contains($"RemoveItem:{serverHandle}:{handle}", server.Operations);
+        }
+
+        Assert.Empty(session.HandleRegistry.ItemHandles);
+    }
+
+    /// <summary>
+    ///     The phased lifecycle: every uncached tag is added and advised before
+    ///     the wait, and none is torn down until the wait is over.
+    /// </summary>
+    [Fact]
+    public void ReadBulk_AddsAndAdvisesEverySnapshotTagBeforeTearingAnyDown()
+    {
+        RecordingMxAccessServer server = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink());
+        int serverHandle = session.Register("client");
+
+        session.ReadBulk(serverHandle, new[] { "Tag.A", "Tag.B", "Tag.C" }, TimeSpan.FromMilliseconds(50), () => { });
+
+        List<string> lifecycle = server.Operations.Where(op => !op.StartsWith("Register", StringComparison.Ordinal)).ToList();
+        int lastAdvise = lifecycle.FindLastIndex(op => op.StartsWith("Advise:", StringComparison.Ordinal));
+        int firstUnAdvise = lifecycle.FindIndex(op => op.StartsWith("UnAdvise:", StringComparison.Ordinal));
+        int firstRemove = lifecycle.FindIndex(op => op.StartsWith("RemoveItem:", StringComparison.Ordinal));
+        Assert.Equal(3, lifecycle.Count(op => op.StartsWith("AddItem:", StringComparison.Ordinal)));
+        Assert.Equal(3, lifecycle.Count(op => op.StartsWith("Advise:", StringComparison.Ordinal)));
+        Assert.True(lastAdvise < firstUnAdvise, string.Join(",", lifecycle));
+        Assert.True(lastAdvise < firstRemove, string.Join(",", lifecycle));
+    }
+
+    /// <summary>
+    ///     Every per-tag outcome keeps its exact pre-batching shape and request
+    ///     order in one mixed request: blank address, cached-and-advised fast
+    ///     path, a tag that delivers, an AddItem failure, an Advise failure, and
+    ///     a tag that never delivers. A delivering tag returns as soon as its
+    ///     value lands rather than waiting out the timeout for the others.
+    /// </summary>
+    [Fact]
+    public void ReadBulk_MixedOutcomes_PreservesPerTagResultsAndOrder()
+    {
+        RecordingMxAccessServer server = new();
+        server.FailAddItemFor.Add("Tag.AddFails");
+        server.FailAdviseFor.Add("Tag.AdviseFails");
+        MxAccessValueCache valueCache = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink(), valueCache: valueCache);
+        int serverHandle = session.Register("client");
+
+        // Caller-owned advised tag with a cached value: served from the cache.
+        int cachedHandle = session.AddItem(serverHandle, "Tag.Cached");
+        session.Advise(serverHandle, cachedHandle);
+        valueCache.Set(serverHandle, cachedHandle, BuildDataChange(serverHandle, cachedHandle, 11));
+        int operationsBeforeRead = server.Operations.Count;
+
+        // Pump step: deliver Tag.Delivers' first OnDataChange once advised.
+        Action pumpStep = () =>
+        {
+            if (server.TryGetAdvisedHandle("Tag.Delivers", out int handle)
+                && !valueCache.TryGet(serverHandle, handle, out _))
+            {
+                valueCache.Set(serverHandle, handle, BuildDataChange(serverHandle, handle, 42));
+            }
+        };
+
+        string[] tags = { " ", "Tag.Cached", "Tag.Delivers", "Tag.AddFails", "Tag.AdviseFails", "Tag.Silent" };
+        IReadOnlyList<BulkReadResult> results = session.ReadBulk(serverHandle, tags, TimeSpan.FromMilliseconds(150), pumpStep);
+
+        Assert.Equal(tags.Length, results.Count);
+        for (int index = 0; index < tags.Length; index++)
+        {
+            Assert.Equal(tags[index], results[index].TagAddress);
+        }
+
+        Assert.False(results[0].WasSuccessful);
+        Assert.Equal(0, results[0].ItemHandle);
+        Assert.Equal("Tag address is required.", results[0].ErrorMessage);
+
+        Assert.True(results[1].WasSuccessful);
+        Assert.True(results[1].WasCached);
+        Assert.Equal(cachedHandle, results[1].ItemHandle);
+        Assert.Equal(11, results[1].Value.Int32Value);
+
+        Assert.True(results[2].WasSuccessful);
+        Assert.False(results[2].WasCached);
+        Assert.Equal(server.HandleForTag("Tag.Delivers"), results[2].ItemHandle);
+        Assert.Equal(42, results[2].Value.Int32Value);
+        Assert.Equal(192, results[2].Quality);
+
+        Assert.False(results[3].WasSuccessful);
+        Assert.Equal(0, results[3].ItemHandle);
+        Assert.Equal("AddItem failed for Tag.AddFails", results[3].ErrorMessage);
+
+        Assert.False(results[4].WasSuccessful);
+        Assert.Equal(server.HandleForTag("Tag.AdviseFails"), results[4].ItemHandle);
+        Assert.Equal("Advise failed for Tag.AdviseFails", results[4].ErrorMessage);
+
+        Assert.False(results[5].WasSuccessful);
+        Assert.Equal(server.HandleForTag("Tag.Silent"), results[5].ItemHandle);
+        Assert.Contains(ReadBulkTimeoutText, results[5].ErrorMessage, StringComparison.OrdinalIgnoreCase);
+
+        // The caller's subscription is untouched; the snapshot handles are gone
+        // (the Advise-failed one removed without an UnAdvise).
+        List<string> readOperations = server.Operations.Skip(operationsBeforeRead).ToList();
+        Assert.DoesNotContain($"UnAdvise:{serverHandle}:{cachedHandle}", readOperations);
+        Assert.DoesNotContain($"RemoveItem:{serverHandle}:{cachedHandle}", readOperations);
+        Assert.DoesNotContain($"UnAdvise:{serverHandle}:{server.HandleForTag("Tag.AdviseFails")}", readOperations);
+        Assert.Contains($"RemoveItem:{serverHandle}:{server.HandleForTag("Tag.AdviseFails")}", readOperations);
+        Assert.Contains($"UnAdvise:{serverHandle}:{server.HandleForTag("Tag.Delivers")}", readOperations);
+        Assert.Contains($"RemoveItem:{serverHandle}:{server.HandleForTag("Tag.Silent")}", readOperations);
+        RegisteredItemHandle remaining = Assert.Single(session.HandleRegistry.ItemHandles);
+        Assert.Equal(cachedHandle, remaining.ItemHandle);
+    }
+
+    /// <summary>
+    ///     When every snapshot tag delivers, the wait ends as soon as the last
+    ///     value lands instead of running out the deadline.
+    /// </summary>
+    [Fact]
+    public void ReadBulk_AllTagsDeliver_ReturnsWithoutWaitingOutTheTimeout()
+    {
+        RecordingMxAccessServer server = new();
+        MxAccessValueCache valueCache = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink(), valueCache: valueCache);
+        int serverHandle = session.Register("client");
+        string[] tags = { "Tag.1", "Tag.2", "Tag.3" };
+
+        Action pumpStep = () =>
+        {
+            foreach (string tag in tags)
+            {
+                if (server.TryGetAdvisedHandle(tag, out int handle) && !valueCache.TryGet(serverHandle, handle, out _))
+                {
+                    valueCache.Set(serverHandle, handle, BuildDataChange(serverHandle, handle, handle));
+                }
+            }
+        };
+
+        Stopwatch stopwatch = Stopwatch.StartNew();
+        IReadOnlyList<BulkReadResult> results = session.ReadBulk(serverHandle, tags, TimeSpan.FromSeconds(10), pumpStep);
+        stopwatch.Stop();
+
+        Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(5), $"ReadBulk took {stopwatch.Elapsed.TotalMilliseconds:F0} ms.");
+        Assert.All(results, result => Assert.True(result.WasSuccessful));
+        for (int index = 0; index < tags.Length; index++)
+        {
+            Assert.Equal(results[index].ItemHandle, results[index].Value.Int32Value);
+        }
+    }
+
+    /// <summary>
+    ///     Ghost entry: an OnDataChange/OnWriteComplete already queued when
+    ///     RemoveItem ran is dispatched after the eviction and recreates the
+    ///     entries. When MXAccess later reuses that item id, AddItem must drop
+    ///     them so ReadBulk's cached path cannot serve the stale value.
+    /// </summary>
+    [Fact]
+    public void AddItem_ReusingAnEvictedHandleId_DropsGhostCacheEntries()
+    {
+        RecordingMxAccessServer server = new();
+        CompletionCacheSink sink = new();
+        MxAccessValueCache valueCache = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, sink, valueCache: valueCache);
+        int serverHandle = session.Register("client");
+
+        int first = session.AddItem(serverHandle, "Tag.Old");
+        session.Advise(serverHandle, first);
+        session.RemoveItem(serverHandle, first);
+
+        // Late dispatch of callbacks queued before the RemoveItem.
+        valueCache.Set(serverHandle, first, BuildDataChange(serverHandle, first, 999));
+        RecordCompletion(sink.WriteCompletionCache, serverHandle, first);
+
+        server.NextItemHandleOverride = first;
+        int reused = session.AddItem(serverHandle, "Tag.New");
+        Assert.Equal(first, reused);
+
+        Assert.False(valueCache.TryGet(serverHandle, reused, out _));
+        Assert.Equal(0UL, sink.WriteCompletionCache.CurrentVersion(serverHandle, reused));
+
+        // Once advised, ReadBulk on the new tag does not report the ghost as cached.
+        session.Advise(serverHandle, reused);
+        BulkReadResult result = Assert.Single(session.ReadBulk(serverHandle, new[] { "Tag.New" }, TimeSpan.FromMilliseconds(30), () => { }));
+        Assert.False(result.WasCached);
+        Assert.False(result.WasSuccessful);
+    }
+
+    /// <summary>
+    ///     A handle MXAccess returns while the registry still tracks it is a
+    ///     live item, so its cached value must survive the AddItem.
+    /// </summary>
+    [Fact]
+    public void AddItem_ReturningAStillRegisteredHandle_KeepsItsCachedValue()
+    {
+        RecordingMxAccessServer server = new() { ReuseHandleForSameTag = true };
+        MxAccessValueCache valueCache = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink(), valueCache: valueCache);
+        int serverHandle = session.Register("client");
+        int handle = session.AddItem(serverHandle, "Tag.Live");
+        valueCache.Set(serverHandle, handle, BuildDataChange(serverHandle, handle, 5));
+
+        Assert.Equal(handle, session.AddItem(serverHandle, "Tag.Live"));
+
+        Assert.True(valueCache.TryGet(serverHandle, handle, out _));
+    }
+
+    /// <summary>
+    ///     Duplicate addresses that MXAccess resolves to one handle: the handle
+    ///     is advised, unadvised, and removed exactly once, and every duplicate
+    ///     receives the shared outcome.
+    /// </summary>
+    [Fact]
+    public void ReadBulk_DuplicateTagsResolvingToOneHandle_AdviseAndTearDownOnceAndFanOutResult()
+    {
+        RecordingMxAccessServer server = new() { ReuseHandleForSameTag = true };
+        MxAccessValueCache valueCache = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink(), valueCache: valueCache);
+        int serverHandle = session.Register("client");
+
+        Action pumpStep = () =>
+        {
+            if (server.TryGetAdvisedHandle("Tag.Dup", out int handle) && !valueCache.TryGet(serverHandle, handle, out _))
+            {
+                valueCache.Set(serverHandle, handle, BuildDataChange(serverHandle, handle, 77));
+            }
+        };
+
+        IReadOnlyList<BulkReadResult> results = session.ReadBulk(
+            serverHandle,
+            new[] { "Tag.Dup", "Tag.Other", "Tag.Dup" },
+            TimeSpan.FromMilliseconds(100),
+            pumpStep);
+
+        int dupHandle = server.HandleForTag("Tag.Dup");
+        string advise = "Advise:" + serverHandle + ":" + dupHandle;
+        string unAdvise = "UnAdvise:" + serverHandle + ":" + dupHandle;
+        string remove = "RemoveItem:" + serverHandle + ":" + dupHandle;
+        Assert.True(results[0].WasSuccessful);
+        Assert.True(results[2].WasSuccessful);
+        Assert.Equal(dupHandle, results[0].ItemHandle);
+        Assert.Equal(dupHandle, results[2].ItemHandle);
+        Assert.Equal(77, results[2].Value.Int32Value);
+        Assert.Equal("Tag.Dup", results[2].TagAddress);
+        Assert.False(results[1].WasSuccessful);
+        Assert.Equal("Tag.Other", results[1].TagAddress);
+
+        Assert.Equal(1, server.Operations.Count(op => op == advise));
+        Assert.Equal(1, server.Operations.Count(op => op == unAdvise));
+        Assert.Equal(1, server.Operations.Count(op => op == remove));
+        Assert.Empty(session.HandleRegistry.ItemHandles);
+    }
+
+    /// <summary>
+    ///     Large batches pump during the add/advise and teardown phases too,
+    ///     not only inside the wait, so the STA heartbeat stays fresh.
+    /// </summary>
+    [Fact]
+    public void ReadBulk_LargeBatch_PumpsDuringAddAdviseAndTeardownPhases()
+    {
+        RecordingMxAccessServer server = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(server, new NoopEventSink());
+        int serverHandle = session.Register("client");
+        string[] tags = Enumerable.Range(0, 3 * MxAccessSession.ReadBulkPumpInterval).Select(i => "Tag." + i).ToArray();
+        List<int> pumpAtOperation = new();
+
+        session.ReadBulk(serverHandle, tags, TimeSpan.FromMilliseconds(20), () => pumpAtOperation.Add(server.Operations.Count));
+
+        int lastAdvise = server.Operations.FindLastIndex(op => op.StartsWith("Advise:", StringComparison.Ordinal));
+        int firstUnAdvise = server.Operations.FindIndex(op => op.StartsWith("UnAdvise:", StringComparison.Ordinal));
+        Assert.Contains(pumpAtOperation, at => at < lastAdvise);
+        Assert.Contains(pumpAtOperation, at => at > firstUnAdvise);
+    }
+
+    private static void RecordCompletion(MxAccessWriteCompletionCache cache, int serverHandle, int itemHandle)
+    {
+        RepeatedField<MxStatusProxy> statuses = new() { new MxStatusProxy { Category = MxStatusCategory.Ok } };
+        cache.Record(serverHandle, itemHandle, statuses);
+    }
+
+    private static MxEvent BuildDataChange(int serverHandle, int itemHandle, int value)
+    {
+        MxEvent mxEvent = new()
+        {
+            Family = MxEventFamily.OnDataChange,
+            ServerHandle = serverHandle,
+            ItemHandle = itemHandle,
+            Quality = 192,
+            SourceTimestamp = Timestamp.FromDateTime(new DateTime(2026, 10, 22, 0, 0, 0, DateTimeKind.Utc)),
+            Value = new MxValue
+            {
+                DataType = MxDataType.Integer,
+                VariantType = "VT_I4",
+                Int32Value = value,
+            },
+            OnDataChange = new OnDataChangeEvent(),
+        };
+        mxEvent.Statuses.Add(new MxStatusProxy { Category = MxStatusCategory.Ok });
+        return mxEvent;
+    }
+
+    /// <summary>Test sink that owns a real completion cache (Attach/Detach are no-ops).</summary>
+    private sealed class CompletionCacheSink : IMxAccessEventSink, IWriteCompletionCacheProvider
+    {
+        /// <inheritdoc />
+        public MxAccessWriteCompletionCache WriteCompletionCache { get; } = new();
+
+        /// <inheritdoc />
+        public void Attach(object mxAccessComObject, string sessionId)
+        {
+        }
+
+        /// <inheritdoc />
+        public void Detach()
+        {
+        }
+    }
+
+    /// <summary>
+    ///     Fake server that hands out increasing handles, records every
+    ///     lifecycle call, and can fail AddItem or Advise for named tags.
+    /// </summary>
+    private sealed class RecordingMxAccessServer : IMxAccessServer
+    {
+        private readonly NoopMxAccessServer noop = new();
+        private readonly Dictionary<string, int> handleByTag = new(StringComparer.Ordinal);
+        private readonly Dictionary<int, string> tagByHandle = new();
+        private readonly HashSet<int> advised = new();
+        private int nextServerHandle = 10;
+        private int nextItemHandle = 100;
+
+        public List<string> Operations { get; } = new();
+
+        public HashSet<string> FailAddItemFor { get; } = new(StringComparer.Ordinal);
+
+        public HashSet<string> FailAdviseFor { get; } = new(StringComparer.Ordinal);
+
+        /// <summary>When set, the next AddItem returns this handle (simulates MXAccess reusing an id).</summary>
+        public int? NextItemHandleOverride { get; set; }
+
+        /// <summary>When true, AddItem of an already-added address returns the existing handle.</summary>
+        public bool ReuseHandleForSameTag { get; set; }
+
+        public int HandleForTag(string tag) => handleByTag[tag];
+
+        public bool TryGetAdvisedHandle(string tag, out int handle)
+        {
+            return handleByTag.TryGetValue(tag, out handle) && advised.Contains(handle);
+        }
+
+        public int Register(string clientName)
+        {
+            Operations.Add($"Register:{clientName}");
+            return nextServerHandle++;
+        }
+
+        public void Unregister(int serverHandle) => Operations.Add($"Unregister:{serverHandle}");
+
+        public int AddItem(int serverHandle, string itemDefinition)
+        {
+            Operations.Add($"AddItem:{serverHandle}:{itemDefinition}");
+            if (FailAddItemFor.Contains(itemDefinition))
+            {
+                throw new InvalidOperationException($"AddItem failed for {itemDefinition}");
+            }
+
+            int handle;
+            if (NextItemHandleOverride is int forced)
+            {
+                NextItemHandleOverride = null;
+                handle = forced;
+            }
+            else if (ReuseHandleForSameTag && handleByTag.TryGetValue(itemDefinition, out int existing))
+            {
+                handle = existing;
+            }
+            else
+            {
+                handle = nextItemHandle++;
+            }
+
+            handleByTag[itemDefinition] = handle;
+            tagByHandle[handle] = itemDefinition;
+            return handle;
+        }
+
+        public int AddItem2(int serverHandle, string itemDefinition, string itemContext) => AddItem(serverHandle, itemDefinition);
+
+        public void RemoveItem(int serverHandle, int itemHandle) => Operations.Add($"RemoveItem:{serverHandle}:{itemHandle}");
+
+        public void Advise(int serverHandle, int itemHandle)
+        {
+            Operations.Add($"Advise:{serverHandle}:{itemHandle}");
+            if (tagByHandle.TryGetValue(itemHandle, out string? tag) && FailAdviseFor.Contains(tag))
+            {
+                throw new InvalidOperationException($"Advise failed for {tag}");
+            }
+
+            advised.Add(itemHandle);
+        }
+
+        public void UnAdvise(int serverHandle, int itemHandle)
+        {
+            Operations.Add($"UnAdvise:{serverHandle}:{itemHandle}");
+            advised.Remove(itemHandle);
+        }
+
+        public void AdviseSupervisory(int serverHandle, int itemHandle) => Advise(serverHandle, itemHandle);
+
+        public int AddBufferedItem(int serverHandle, string itemDefinition, string itemContext) => AddItem(serverHandle, itemDefinition);
+
+        public void SetBufferedUpdateInterval(int serverHandle, int updateIntervalMilliseconds) =>
+            noop.SetBufferedUpdateInterval(serverHandle, updateIntervalMilliseconds);
+
+        public object Suspend(int serverHandle, int itemHandle) => noop.Suspend(serverHandle, itemHandle);
+
+        public object Activate(int serverHandle, int itemHandle) => noop.Activate(serverHandle, itemHandle);
+
+        public void Write(int serverHandle, int itemHandle, object? value, int userId) =>
+            Operations.Add($"Write:{serverHandle}:{itemHandle}");
+
+        public void Write2(int serverHandle, int itemHandle, object? value, object? timestampValue, int userId) =>
+            Operations.Add($"Write2:{serverHandle}:{itemHandle}");
+
+        public void WriteSecured(int serverHandle, int itemHandle, int currentUserId, int verifierUserId, object? value) =>
+            Operations.Add($"WriteSecured:{serverHandle}:{itemHandle}");
+
+        public void WriteSecured2(int serverHandle, int itemHandle, int currentUserId, int verifierUserId, object? value, object? timestampValue) =>
+            Operations.Add($"WriteSecured2:{serverHandle}:{itemHandle}");
+
+        public int AuthenticateUser(int serverHandle, string verifyUser, string verifyUserPassword) => 0;
+
+        public int ArchestrAUserToId(int serverHandle, string userIdGuid) => 0;
+    }
+}

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

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Runtime.InteropServices;
 using System.Threading;
@@ -573,6 +574,181 @@ public sealed class MxAccessStaSessionTests
             "Expected StaCallInProgress to clear once the alarm poll returned.");
     }
 
+    /// <summary>
+    /// A transient PollOnce failure must not fault the whole session: with a
+    /// failure threshold of 3, a poll pattern that never fails three times in
+    /// a row (fail, fail, succeed, ...) keeps the session healthy, and each
+    /// transient failure is logged with type/HRESULT only — never the
+    /// exception message.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RunAlarmPollLoop_TransientFailuresBelowThreshold_DoNotFaultSession()
+    {
+        FakeAlarmCommandHandler handler = new()
+        {
+            PollExceptionFactory = call => call % 3 == 0
+                ? null
+                : new COMException("secret-tag-name GetXmlCurrentAlarms2 failed.", unchecked((int)0x80004005)),
+        };
+        ConcurrentQueue<string> warnings = new();
+        using StaRuntime runtime = CreateRuntime();
+        MxAccessEventQueue eventQueue = new();
+        using MxAccessStaSession session = new(
+            runtime,
+            new FakeMxAccessComObjectFactory(),
+            new FakeMxAccessEventSink(),
+            eventQueue,
+            (_eq, _affinity, _comFactory) => handler);
+        session.AlarmPollInterval = TimeSpan.FromMilliseconds(10);
+        session.AlarmPollFailureThreshold = 3;
+        session.AlarmPollWarningSink = warnings.Enqueue;
+
+        await session.StartAsync("session-1", workerProcessId: 1);
+
+        DateTime deadline = DateTime.UtcNow.AddSeconds(10);
+        while (handler.PollCount < 12 && !eventQueue.IsFaulted && DateTime.UtcNow < deadline)
+        {
+            await Task.Delay(10);
+        }
+
+        Assert.False(eventQueue.IsFaulted, "Transient alarm-poll failures below the threshold must not fault the session.");
+        Assert.True(handler.PollCount >= 12, $"Expected the poll loop to keep polling; saw {handler.PollCount} polls.");
+        Assert.NotEmpty(warnings);
+        Assert.All(warnings, line =>
+        {
+            Assert.Contains("event=AlarmPollFailed", line, StringComparison.Ordinal);
+            Assert.Contains("hresult=0x80004005", line, StringComparison.Ordinal);
+            Assert.DoesNotContain("secret-tag-name", line, StringComparison.Ordinal);
+        });
+    }
+
+    /// <summary>
+    /// A run of failures reaching the threshold with no success in between
+    /// faults the session exactly once and stops the loop: two warnings, then
+    /// the fault on the third consecutive failure, then no further polls.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RunAlarmPollLoop_ConsecutiveFailuresReachThreshold_FaultsAndStops()
+    {
+        FakeAlarmCommandHandler handler = new()
+        {
+            PollException = new COMException("GetXmlCurrentAlarms2 failed.", unchecked((int)0x80004005)),
+        };
+        ConcurrentQueue<string> warnings = new();
+        using StaRuntime runtime = CreateRuntime();
+        MxAccessEventQueue eventQueue = new();
+        using MxAccessStaSession session = new(
+            runtime,
+            new FakeMxAccessComObjectFactory(),
+            new FakeMxAccessEventSink(),
+            eventQueue,
+            (_eq, _affinity, _comFactory) => handler);
+        session.AlarmPollInterval = TimeSpan.FromMilliseconds(10);
+        session.AlarmPollFailureThreshold = 3;
+        session.AlarmPollWarningSink = warnings.Enqueue;
+
+        await session.StartAsync("session-1", workerProcessId: 1);
+
+        DateTime deadline = DateTime.UtcNow.AddSeconds(10);
+        while (!eventQueue.IsFaulted && DateTime.UtcNow < deadline)
+        {
+            await Task.Delay(10);
+        }
+
+        Assert.True(eventQueue.IsFaulted);
+        await Task.Delay(200);
+        Assert.Equal(3, handler.PollCount);
+        Assert.Equal(2, warnings.Count);
+        WorkerFault? fault = session.DrainFault();
+        Assert.NotNull(fault);
+        Assert.Equal(WorkerFaultCategory.MxaccessEventConversionFailed, fault!.Category);
+    }
+
+    /// <summary>
+    /// WRK-02: when the STA loop dies after the session started, the death is
+    /// recorded as a StaHung fault on the event queue — the path the IPC drain
+    /// loop turns into a WorkerFault frame and a terminated pipe session.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StaLoopDeath_AfterStart_RecordsStaHungFaultOnEventQueue()
+    {
+        ControllableMessagePump pump = new();
+        using StaRuntime runtime = new(new NoopComApartmentInitializer(), pump, TimeSpan.FromMilliseconds(25));
+        MxAccessEventQueue eventQueue = new();
+        using MxAccessStaSession session = new(
+            runtime,
+            new FakeMxAccessComObjectFactory(),
+            new FakeMxAccessEventSink(),
+            eventQueue);
+
+        await session.StartAsync("session-1", workerProcessId: 1);
+        Assert.False(eventQueue.IsFaulted);
+
+        pump.ThrowOnNextPump(new InvalidOperationException("secret-tag-name pump failed."));
+
+        DateTime deadline = DateTime.UtcNow.AddSeconds(5);
+        while (!eventQueue.IsFaulted && DateTime.UtcNow < deadline)
+        {
+            await Task.Delay(10);
+        }
+
+        Assert.True(eventQueue.IsFaulted, "Expected the STA loop death to fault the event queue.");
+        WorkerFault? fault = session.DrainFault();
+        Assert.NotNull(fault);
+        Assert.Equal(WorkerFaultCategory.StaHung, fault!.Category);
+        Assert.Equal(typeof(InvalidOperationException).FullName, fault.ExceptionType);
+        Assert.Contains("STA loop terminated", fault.DiagnosticMessage, StringComparison.Ordinal);
+        Assert.Contains("exceptionType=System.InvalidOperationException", fault.DiagnosticMessage, StringComparison.Ordinal);
+        Assert.Contains("hresult=0x", fault.DiagnosticMessage, StringComparison.Ordinal);
+        Assert.DoesNotContain("secret-tag-name", fault.DiagnosticMessage, StringComparison.Ordinal);
+        Assert.DoesNotContain("secret-tag-name", fault.ProtocolStatus.Message, StringComparison.Ordinal);
+        Assert.Equal(ProtocolStatusCode.WorkerUnavailable, fault.ProtocolStatus.Code);
+    }
+
+    /// <summary>
+    /// After the STA loop dies, the alarm poll's fast-failed InvokeAsync must
+    /// not record a second, mislabelled alarm-poll fault: once the StaHung
+    /// fault is drained, nothing else is recorded.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RunAlarmPollLoop_AfterStaLoopFault_DoesNotRecordAlarmPollFault()
+    {
+        ControllableMessagePump pump = new();
+        FakeAlarmCommandHandler handler = new();
+        using StaRuntime runtime = new(new NoopComApartmentInitializer(), pump, TimeSpan.FromMilliseconds(25));
+        MxAccessEventQueue eventQueue = new();
+        using MxAccessStaSession session = new(
+            runtime,
+            new FakeMxAccessComObjectFactory(),
+            new FakeMxAccessEventSink(),
+            eventQueue,
+            (_eq, _affinity, _comFactory) => handler);
+        session.AlarmPollInterval = TimeSpan.FromMilliseconds(10);
+        session.AlarmPollFailureThreshold = 1;
+        session.AlarmPollWarningSink = _ => { };
+
+        await session.StartAsync("session-1", workerProcessId: 1);
+        pump.ThrowOnNextPump(new InvalidOperationException("pump failed"));
+
+        DateTime deadline = DateTime.UtcNow.AddSeconds(5);
+        while (!eventQueue.IsFaulted && DateTime.UtcNow < deadline)
+        {
+            await Task.Delay(10);
+        }
+
+        WorkerFault? first = session.DrainFault();
+        Assert.NotNull(first);
+        Assert.Equal(WorkerFaultCategory.StaHung, first!.Category);
+
+        // Give the poll loop several ticks to (wrongly) record its own fault.
+        await Task.Delay(300);
+        Assert.Null(session.DrainFault());
+    }
+
     /// <summary>
     /// The STA-affinity guard throws when an
     /// IMxAccessAlarmConsumer call is attempted off the thread that created
@@ -621,6 +797,9 @@ public sealed class MxAccessStaSessionTests
         /// <summary>Exception thrown by PollOnce; null to succeed.</summary>
         public Exception? PollException { get; set; }
 
+        /// <summary>Per-call exception factory (1-based call number); null result succeeds.</summary>
+        public Func<int, Exception?>? PollExceptionFactory { get; set; }
+
         /// <summary>When set, <see cref="PollOnce"/> blocks until <see cref="ReleasePoll"/> is called.</summary>
         public bool BlockPoll { get; set; }
 
@@ -699,6 +878,22 @@ public sealed class MxAccessStaSessionTests
             {
                 throw PollException;
             }
+
+            Func<int, Exception?>? factory = PollExceptionFactory;
+            if (factory is not null)
+            {
+                int callNumber;
+                lock (gate)
+                {
+                    callNumber = pollCount;
+                }
+
+                Exception? exception = factory(callNumber);
+                if (exception is not null)
+                {
+                    throw exception;
+                }
+            }
         }
 
         /// <inheritdoc />

src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs

@@ -2,6 +2,7 @@ using System;
 using System.Threading;
 using System.Threading.Tasks;
 using ZB.MOM.WW.MxGateway.Worker.Sta;
+using ZB.MOM.WW.MxGateway.Worker.Tests.TestSupport;
 
 namespace ZB.MOM.WW.MxGateway.Worker.Tests.Sta;
 
@@ -155,6 +156,145 @@ public sealed class StaRuntimeTests
         Assert.Contains("shutting down", exception.Message);
     }
 
+    /// <summary>
+    ///     A backlog of queued work items must not starve inbound COM
+    ///     callbacks: the loop pumps Windows messages between items rather
+    ///     than only after draining the whole queue. A gate item holds the STA
+    ///     while two more items queue behind it; once released, the second
+    ///     queued item must observe a pump that happened after the first.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ProcessQueuedCommands_PumpsMessagesBetweenQueuedWorkItems()
+    {
+        ControllableMessagePump pump = new();
+        using StaRuntime runtime = new(new RecordingComApartmentInitializer(), pump, TimeSpan.FromSeconds(30));
+        runtime.Start();
+
+        using ManualResetEventSlim gateEntered = new(false);
+        using ManualResetEventSlim releaseGate = new(false);
+        Task gate = runtime.InvokeAsync(() =>
+        {
+            gateEntered.Set();
+            releaseGate.Wait(TimeSpan.FromSeconds(10));
+        });
+        Assert.True(gateEntered.Wait(TimeSpan.FromSeconds(5)));
+
+        Task<int> first = runtime.InvokeAsync(() => pump.PumpCount);
+        Task<int> second = runtime.InvokeAsync(() => pump.PumpCount);
+        releaseGate.Set();
+
+        await gate;
+        int pumpsSeenByFirst = await first;
+        int pumpsSeenBySecond = await second;
+
+        Assert.True(
+            pumpsSeenBySecond > pumpsSeenByFirst,
+            $"Expected a pump between queued work items (first saw {pumpsSeenByFirst}, second saw {pumpsSeenBySecond}).");
+    }
+
+    /// <summary>
+    ///     WRK-02: an exception that kills the STA loop after a successful
+    ///     <see cref="StaRuntime.Start"/> must be surfaced — recorded on
+    ///     <see cref="StaRuntime.LoopFault"/>, handed to the registered fault
+    ///     handler, and turned into fast failures for every later
+    ///     <see cref="StaRuntime.InvokeAsync{T}(Func{T}, CancellationToken)"/> —
+    ///     rather than parked in a startup field nothing reads.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task LoopException_AfterStart_IsReportedToHandlerAndFailsLaterCommands()
+    {
+        ControllableMessagePump pump = new();
+        RecordingComApartmentInitializer initializer = new();
+        using StaRuntime runtime = new(initializer, pump, TimeSpan.FromMilliseconds(25));
+        runtime.Start();
+
+        TaskCompletionSource<Exception> reported = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        runtime.RegisterLoopFaultHandler(exception => reported.TrySetResult(exception));
+        Assert.Null(runtime.LoopFault);
+
+        InvalidOperationException loopException = new("pump failed");
+        pump.ThrowOnNextPump(loopException);
+
+        Task completed = await Task.WhenAny(reported.Task, Task.Delay(TimeSpan.FromSeconds(5)));
+        Assert.Same(reported.Task, completed);
+        Assert.Same(loopException, await reported.Task);
+        Assert.Same(loopException, runtime.LoopFault);
+
+        InvalidOperationException rejected = await Assert.ThrowsAsync<InvalidOperationException>(
+            () => runtime.InvokeAsync(() => 1));
+        Assert.IsNotType<StaRuntimeShutdownException>(rejected);
+        Assert.Same(loopException, rejected.InnerException);
+
+        // The dying thread still runs its apartment cleanup.
+        Assert.True(runtime.Shutdown(TimeSpan.FromSeconds(5)));
+        Assert.False(runtime.IsRunning);
+        Assert.Equal(1, initializer.UninitializeCount);
+    }
+
+    /// <summary>
+    ///     A handler registered after the loop already died still receives the
+    ///     fault, so the window between Start and registration loses nothing.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RegisterLoopFaultHandler_AfterLoopDied_InvokesHandlerImmediately()
+    {
+        ControllableMessagePump pump = new();
+        using StaRuntime runtime = new(new RecordingComApartmentInitializer(), pump, TimeSpan.FromMilliseconds(25));
+        runtime.Start();
+        InvalidOperationException loopException = new("pump failed");
+        pump.ThrowOnNextPump(loopException);
+
+        DateTime deadline = DateTime.UtcNow.AddSeconds(5);
+        while (runtime.LoopFault is null && DateTime.UtcNow < deadline)
+        {
+            await Task.Delay(10);
+        }
+
+        Exception? seen = null;
+        runtime.RegisterLoopFaultHandler(exception => seen = exception);
+
+        Assert.Same(loopException, seen);
+    }
+
+    /// <summary>A requested shutdown is not a loop fault.</summary>
+    [Fact]
+    public void Shutdown_DoesNotRecordLoopFault()
+    {
+        using StaRuntime runtime = new(new RecordingComApartmentInitializer(), new ControllableMessagePump(), TimeSpan.FromMilliseconds(25));
+        runtime.Start();
+        bool invoked = false;
+        runtime.RegisterLoopFaultHandler(_ => invoked = true);
+
+        Assert.True(runtime.Shutdown(TimeSpan.FromSeconds(5)));
+
+        Assert.Null(runtime.LoopFault);
+        Assert.False(invoked);
+    }
+
+    /// <summary>
+    ///     A loop that throws while a requested shutdown is in progress is
+    ///     being torn down on purpose: the loop-fault handler must not run.
+    /// </summary>
+    [Fact]
+    public void LoopException_DuringRequestedShutdown_DoesNotInvokeHandler()
+    {
+        ControllableMessagePump pump = new();
+        using StaRuntime runtime = new(new RecordingComApartmentInitializer(), pump, TimeSpan.FromSeconds(30));
+        runtime.Start();
+        bool invoked = false;
+        runtime.RegisterLoopFaultHandler(_ => invoked = true);
+
+        // The loop is parked in the 30 s idle wait; Shutdown wakes it and the
+        // very next pump throws with shutdownRequested already set.
+        pump.ThrowOnNextPump(new InvalidOperationException("pump failed during shutdown"));
+        Assert.True(runtime.Shutdown(TimeSpan.FromSeconds(5)));
+
+        Assert.False(invoked);
+    }
+
     private static StaRuntime CreateRuntime(RecordingComApartmentInitializer initializer)
     {
         return new StaRuntime(

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

@@ -0,0 +1,47 @@
+using System;
+using System.Threading;
+using ZB.MOM.WW.MxGateway.Worker.Sta;
+
+namespace ZB.MOM.WW.MxGateway.Worker.Tests.TestSupport;
+
+/// <summary>
+/// Test <see cref="IStaMessagePump"/> that counts pump calls and can be armed
+/// to throw from the next pump, so tests can observe the STA loop's pump
+/// interleaving and drive a post-start loop death without a real Win32
+/// message queue. The idle wait blocks on the command wake event exactly
+/// like the production pump's <c>MsgWaitForMultipleObjectsEx</c> does.
+/// </summary>
+internal sealed class ControllableMessagePump : IStaMessagePump
+{
+    private int pumpCount;
+    private Exception? throwOnNextPump;
+
+    /// <summary>Gets the number of <see cref="PumpPendingMessages"/> calls so far.</summary>
+    public int PumpCount => Volatile.Read(ref pumpCount);
+
+    /// <summary>Arms the pump to throw <paramref name="exception"/> from its next pump call.</summary>
+    /// <param name="exception">The exception to throw once.</param>
+    public void ThrowOnNextPump(Exception exception)
+    {
+        Volatile.Write(ref throwOnNextPump, exception);
+    }
+
+    /// <inheritdoc />
+    public void WaitForWorkOrMessages(WaitHandle commandWakeEvent, TimeSpan timeout)
+    {
+        commandWakeEvent.WaitOne(timeout);
+    }
+
+    /// <inheritdoc />
+    public int PumpPendingMessages()
+    {
+        Interlocked.Increment(ref pumpCount);
+        Exception? exception = Interlocked.Exchange(ref throwOnNextPump, null);
+        if (exception is not null)
+        {
+            throw exception;
+        }
+
+        return 0;
+    }
+}

src/ZB.MOM.WW.MxGateway.Worker/Bootstrap/WorkerExitCode.cs

@@ -9,4 +9,13 @@ public enum WorkerExitCode
     MissingNonce = 4,
     PipeConnectionFailed = 5,
     ProtocolViolation = 6,
+
+    /// <summary>
+    ///     An exception escaped to <c>AppDomain.UnhandledException</c> on some
+    ///     thread (a background task, a COM callback thread, a timer) rather
+    ///     than through <c>WorkerApplication.Run</c>'s own catch blocks. The
+    ///     worker logs it and exits with this code instead of the CLR's crash
+    ///     exit code.
+    /// </summary>
+    UnhandledException = 7,
 }

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

@@ -246,13 +246,32 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
         {
             IMxAccessAlarmConsumer primary = consumerFactory()
                 ?? throw new InvalidOperationException("Alarm consumer factory returned null.");
-            IMxAccessAlarmConsumer standby = BuildStandby(watchList);
-            AlarmFailoverConfig? failoverConfig = command.Failover;
-            FailoverSettings settings = new FailoverSettings(
-                failoverConfig?.ConsecutiveFailureThreshold ?? 3,
-                failoverConfig?.FailbackProbeIntervalSeconds ?? 30,
-                failoverConfig?.FailbackStableProbes ?? 3);
-            return new FailoverAlarmConsumer(primary, standby, settings);
+            IMxAccessAlarmConsumer? standby = null;
+            try
+            {
+                standby = BuildStandby(watchList);
+                AlarmFailoverConfig? failoverConfig = command.Failover;
+                FailoverSettings settings = new FailoverSettings(
+                    failoverConfig?.ConsecutiveFailureThreshold ?? 3,
+                    failoverConfig?.FailbackProbeIntervalSeconds ?? 30,
+                    failoverConfig?.FailbackStableProbes ?? 3);
+                return new FailoverAlarmConsumer(primary, standby, settings);
+            }
+            catch
+            {
+                // The primary (and a standby that was built before the
+                // composite constructor threw) is owned by nothing yet: the
+                // failed Subscribe never reaches the dispatcher that would
+                // dispose it, so release it here or its wnwrap COM object
+                // leaks for the life of the worker.
+                if (standby is not null)
+                {
+                    try { standby.Dispose(); } catch { /* swallow — preserve the build failure */ }
+                }
+
+                try { primary.Dispose(); } catch { /* swallow — preserve the build failure */ }
+                throw;
+            }
         }
 
         // Alarmmgr, or Unspecified with an empty watch list — primary only.

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

@@ -49,6 +49,17 @@ public static class AlarmRecordTransitionMapper
     ///         <item><description><c>previous=*Rtn</c> + <c>current=*Alm</c> → Raise (re-trigger after clear).</description></item>
     ///         <item><description>Anything else → Unspecified (no proto kind to emit).</description></item>
     ///     </list>
+    ///     <para>
+    ///         <see cref="AlarmTransitionKind.Retrigger"/> is never returned. It is
+    ///         reserved in the contract, but neither alarm source gives the worker
+    ///         a native re-alarm signal: the alarmmgr XML carries only the four
+    ///         <c>*_ALM</c>/<c>*_RTN</c> states, so a return-to-alarm is
+    ///         indistinguishable from a fresh raise and is reported as Raise, and
+    ///         the subtag source observes no re-alarm counter. Deliberately not
+    ///         synthesized (the gateway forwards only transitions the source
+    ///         reports — see docs/AlarmClientDiscovery.md, "Retrigger
+    ///         transitions").
+    ///     </para>
     /// </summary>
     /// <param name="previous">The previous alarm state kind.</param>
     /// <param name="current">The current alarm state kind.</param>

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

@@ -222,6 +222,13 @@ public sealed class MxAccessSession : IDisposable
 
         mxAccessServer.Unregister(serverHandle);
         handleRegistry.UnregisterServerHandle(serverHandle);
+        // Unregister drops every item under the server in one call, so the
+        // per-item evictions in RemoveItem never run for them. Evict both
+        // per-handle caches by server so a long-lived worker that registers
+        // and unregisters clients does not accumulate one entry per item it
+        // ever wrote or read, and a reused server handle id starts clean.
+        valueCache.RemoveServer(serverHandle);
+        writeCompletionCache.RemoveServer(serverHandle);
     }
 
     /// <summary>Adds an item to an MXAccess server and returns the item handle.</summary>
@@ -235,6 +242,7 @@ public sealed class MxAccessSession : IDisposable
         ThrowIfDisposed();
 
         int itemHandle = mxAccessServer.AddItem(serverHandle, itemDefinition);
+        EvictStaleHandleState(serverHandle, itemHandle);
         handleRegistry.RegisterItemHandle(
             serverHandle,
             itemHandle,
@@ -258,6 +266,7 @@ public sealed class MxAccessSession : IDisposable
         ThrowIfDisposed();
 
         int itemHandle = mxAccessServer.AddItem2(serverHandle, itemDefinition, itemContext);
+        EvictStaleHandleState(serverHandle, itemHandle);
         handleRegistry.RegisterItemHandle(
             serverHandle,
             itemHandle,
@@ -268,6 +277,30 @@ public sealed class MxAccessSession : IDisposable
         return itemHandle;
     }
 
+    /// <summary>
+    ///     Drops cached per-handle state left over from a previous lifetime of
+    ///     a handle id MXAccess just handed out again. RemoveItem and Unregister
+    ///     already evict, but an OnDataChange or OnWriteComplete that was
+    ///     queued on the STA when they ran is dispatched afterwards (the loop
+    ///     pumps between work items) and recreates the entry. Without this, a
+    ///     reused id would let ReadBulk's cached fast path serve that stale
+    ///     value as <c>was_cached = true</c>. A handle the registry still
+    ///     tracks is a live item (MXAccess returned an existing handle), so its
+    ///     current entries are kept.
+    /// </summary>
+    private void EvictStaleHandleState(
+        int serverHandle,
+        int itemHandle)
+    {
+        if (handleRegistry.ContainsItemHandle(serverHandle, itemHandle))
+        {
+            return;
+        }
+
+        valueCache.Remove(serverHandle, itemHandle);
+        writeCompletionCache.Remove(serverHandle, itemHandle);
+    }
+
     /// <summary>Removes an item from an MXAccess server.</summary>
     /// <param name="serverHandle">Handle returned by the worker.</param>
     /// <param name="itemHandle">Handle returned by the worker.</param>
@@ -283,6 +316,11 @@ public sealed class MxAccessSession : IDisposable
         // same handle id (which MXAccess may reuse) does not serve a stale
         // OnDataChange snapshot from the previous lifetime.
         valueCache.Remove(serverHandle, itemHandle);
+        // Same for the OnWriteComplete completion entry: the event sink records
+        // one per written handle pair, so without this an AddItem -> Write ->
+        // RemoveItem churn grows the cache by one entry per cycle for the life
+        // of the worker.
+        writeCompletionCache.Remove(serverHandle, itemHandle);
     }
 
     /// <summary>Advises on item changes with plain subscription.</summary>
@@ -397,6 +435,7 @@ public sealed class MxAccessSession : IDisposable
         ThrowIfDisposed();
 
         int itemHandle = mxAccessServer.AddBufferedItem(serverHandle, itemDefinition, itemContext);
+        EvictStaleHandleState(serverHandle, itemHandle);
         handleRegistry.RegisterItemHandle(
             serverHandle,
             itemHandle,
@@ -870,14 +909,28 @@ public sealed class MxAccessSession : IDisposable
     ///     OnDataChange value if the tag is already advised AND a cached value
     ///     exists (no subscription side effects); otherwise takes the AddItem
     ///     + Advise + wait + UnAdvise + RemoveItem snapshot lifecycle itself.
-    ///     <paramref name="timeout"/> bounds the wait per-tag in the snapshot
-    ///     case; <paramref name="pumpStep"/> is invoked on every poll
-    ///     iteration so the worker's STA can dispatch the incoming MXAccess
-    ///     message that carries the value.
+    ///     <paramref name="timeout"/> bounds the wait for every snapshot tag;
+    ///     <paramref name="pumpStep"/> is invoked on every poll iteration so the
+    ///     worker's STA can dispatch the incoming MXAccess message that carries
+    ///     the value.
     /// </summary>
+    /// <remarks>
+    ///     The snapshot tags are processed in phases rather than one at a time:
+    ///     every cache probe runs first, then every uncached tag is added and
+    ///     advised, then one wait with a single shared deadline collects the
+    ///     first OnDataChange per handle, then every snapshot handle is torn
+    ///     down. A serial lifecycle held the STA for up to N x timeout when N
+    ///     tags never produced a value; the phased form holds it for at most
+    ///     one timeout. Each tag's result is still its own: a tag that never
+    ///     delivered fails with the same timed-out diagnostic and its own item
+    ///     handle, an AddItem or Advise failure fails only that tag, and results
+    ///     keep request order. The cache probes run before any snapshot AddItem
+    ///     so a later tag can never be served from an earlier tag's own
+    ///     snapshot subscription.
+    /// </remarks>
     /// <param name="serverHandle">The MXAccess server handle.</param>
     /// <param name="tagAddresses">The tag addresses to read.</param>
-    /// <param name="timeout">The timeout per tag.</param>
+    /// <param name="timeout">The timeout bounding the shared snapshot wait.</param>
     /// <param name="pumpStep">Action invoked on each poll iteration.</param>
     /// <returns>The per-tag read results, in the order the tags were provided.</returns>
     public IReadOnlyList<BulkReadResult> ReadBulk(
@@ -897,91 +950,225 @@ public sealed class MxAccessSession : IDisposable
             throw new ArgumentNullException(nameof(pumpStep));
         }
 
-        List<BulkReadResult> results = new(tagAddresses.Count);
-        foreach (string? tagAddress in tagAddresses)
+        int tagCount = tagAddresses.Count;
+        BulkReadResult?[] results = new BulkReadResult?[tagCount];
+        List<SnapshotRead> snapshots = new();
+
+        // Phase 1: validation and the cached-and-advised fast path. A tag
+        // already advised by the caller with a cached value is answered without
+        // touching the existing subscription — the caller didn't create it, so
+        // ReadBulk must not tear it down.
+        for (int index = 0; index < tagCount; index++)
         {
+            string? tagAddress = tagAddresses[index];
             if (string.IsNullOrWhiteSpace(tagAddress))
             {
-                results.Add(FailedRead(serverHandle, tagAddress ?? string.Empty, itemHandle: 0, wasCached: false, "Tag address is required."));
+                results[index] = FailedRead(serverHandle, tagAddress ?? string.Empty, itemHandle: 0, wasCached: false, "Tag address is required.");
                 continue;
             }
 
-            results.Add(ReadOneTag(serverHandle, tagAddress, timeout, pumpStep));
+            if (TryGetCachedReadFor(serverHandle, tagAddress!, out int cachedItemHandle, out MxAccessValueCache.CachedValue cachedValue))
+            {
+                results[index] = SucceededRead(serverHandle, tagAddress!, cachedItemHandle, wasCached: true, cachedValue);
+                continue;
+            }
+
+            snapshots.Add(new SnapshotRead(index, tagAddress!));
         }
 
-        return results;
+        if (snapshots.Count > 0)
+        {
+            try
+            {
+                ReadSnapshots(serverHandle, snapshots, results, timeout, pumpStep);
+            }
+            finally
+            {
+                TearDownSnapshots(serverHandle, snapshots, pumpStep);
+            }
+        }
+
+        List<BulkReadResult> ordered = new(tagCount);
+        for (int index = 0; index < tagCount; index++)
+        {
+            ordered.Add(results[index]!);
+        }
+
+        return ordered;
     }
 
-    private BulkReadResult ReadOneTag(
+    /// <summary>
+    ///     Number of snapshot items ReadBulk adds/advises (phase 2) or tears
+    ///     down (phase 4) between pump steps. Those phases make one or two COM
+    ///     calls per tag without waiting, so a large batch would otherwise hold
+    ///     the STA with no pump and no activity mark for its whole length; a
+    ///     pump every <see cref="ReadBulkPumpInterval"/> items keeps inbound
+    ///     callbacks flowing and keeps the STA heartbeat fresh, so a big batch
+    ///     cannot drift toward <c>HeartbeatStuckCeiling</c>.
+    /// </summary>
+    internal const int ReadBulkPumpInterval = 16;
+
+    private void ReadSnapshots(
         int serverHandle,
-        string tagAddress,
+        List<SnapshotRead> snapshots,
+        BulkReadResult?[] results,
         TimeSpan timeout,
         Action pumpStep)
     {
-        // 1. Cached-and-advised fast path: scan the registry for a live item
-        //    handle matching this tag and check whether the value cache has a
-        //    payload for it. If so, return the cached value without touching
-        //    the existing subscription — the caller didn't create it, so
-        //    ReadBulk must not tear it down.
-        if (TryGetCachedReadFor(serverHandle, tagAddress, out int cachedItemHandle, out MxAccessValueCache.CachedValue cachedValue))
+        // Phase 2: reserve our own item handle and advise for every uncached
+        // tag. A per-tag AddItem/Advise failure fails only that tag. When
+        // MXAccess hands back a handle an earlier tag in this batch already
+        // advised (a duplicate address), that tag shares the earlier tag's
+        // subscription instead of advising the handle a second time, and gets
+        // the same outcome fanned out to it after the wait.
+        List<SnapshotRead> waiting = new(snapshots.Count);
+        Dictionary<int, SnapshotRead> advisedByHandle = new();
+        List<SnapshotRead> sharing = new();
+        for (int position = 0; position < snapshots.Count; position++)
         {
-            return SucceededRead(
-                serverHandle,
-                tagAddress,
-                cachedItemHandle,
-                wasCached: true,
-                cachedValue);
+            SnapshotRead snapshot = snapshots[position];
+            try
+            {
+                if (position > 0 && position % ReadBulkPumpInterval == 0)
+                {
+                    pumpStep();
+                }
+
+                snapshot.ItemHandle = AddItem(serverHandle, snapshot.TagAddress);
+                if (advisedByHandle.TryGetValue(snapshot.ItemHandle, out SnapshotRead? owner))
+                {
+                    snapshot.SharedWith = owner;
+                    sharing.Add(snapshot);
+                    continue;
+                }
+
+                snapshot.Baseline = valueCache.CurrentVersion(serverHandle, snapshot.ItemHandle);
+                Advise(serverHandle, snapshot.ItemHandle);
+                snapshot.Advised = true;
+                advisedByHandle[snapshot.ItemHandle] = snapshot;
+                waiting.Add(snapshot);
+            }
+            catch (Exception exception)
+            {
+                results[snapshot.Index] = FailedRead(serverHandle, snapshot.TagAddress, snapshot.ItemHandle, wasCached: false, exception.Message);
+            }
         }
 
-        // 2. Snapshot lifecycle. Reserve our own item handle, advise, pump
-        //    until we see a fresh OnDataChange (or the deadline elapses),
-        //    then tear it down.
-        int itemHandle = 0;
-        bool advised = false;
+        if (waiting.Count > 0)
+        {
+            WaitForSnapshots(serverHandle, waiting, results, timeout, pumpStep);
+        }
+
+        foreach (SnapshotRead duplicate in sharing)
+        {
+            BulkReadResult? ownerResult = results[duplicate.SharedWith!.Index];
+            if (ownerResult is null)
+            {
+                continue;
+            }
+
+            BulkReadResult fanned = ownerResult.Clone();
+            fanned.TagAddress = duplicate.TagAddress;
+            results[duplicate.Index] = fanned;
+        }
+    }
+
+    private void WaitForSnapshots(
+        int serverHandle,
+        List<SnapshotRead> waiting,
+        BulkReadResult?[] results,
+        TimeSpan timeout,
+        Action pumpStep)
+    {
+        // Phase 3: one shared deadline. Pump until every advised handle has
+        // seen its first OnDataChange since its baseline, or the deadline
+        // elapses.
+        int[] itemHandles = new int[waiting.Count];
+        ulong[] baselines = new ulong[waiting.Count];
+        for (int index = 0; index < waiting.Count; index++)
+        {
+            itemHandles[index] = waiting[index].ItemHandle;
+            baselines[index] = waiting[index].Baseline;
+        }
+
+        bool[] received = new bool[waiting.Count];
+        MxAccessValueCache.CachedValue[] values = new MxAccessValueCache.CachedValue[waiting.Count];
         try
         {
-            itemHandle = AddItem(serverHandle, tagAddress);
-            ulong baseline = valueCache.CurrentVersion(serverHandle, itemHandle);
-            Advise(serverHandle, itemHandle);
-            advised = true;
-
             DateTime deadline = DateTime.UtcNow + timeout;
-            bool gotValue = valueCache.TryWaitForUpdate(
-                serverHandle,
-                itemHandle,
-                baseline,
-                deadline,
-                pumpStep,
-                out MxAccessValueCache.CachedValue snapshot);
-
-            return gotValue
-                ? SucceededRead(serverHandle, tagAddress, itemHandle, wasCached: false, snapshot)
-                : FailedRead(
-                    serverHandle,
-                    tagAddress,
-                    itemHandle,
-                    wasCached: false,
-                    $"ReadBulk timed out after {timeout.TotalMilliseconds:F0} ms waiting for first OnDataChange.");
+            valueCache.TryWaitForUpdates(serverHandle, itemHandles, baselines, deadline, pumpStep, received, values);
         }
         catch (Exception exception)
         {
-            return FailedRead(serverHandle, tagAddress, itemHandle, wasCached: false, exception.Message);
-        }
-        finally
-        {
-            // Snapshot teardown — best-effort. Errors here are noted on the
-            // diagnostic message of the original result (above) by appending
-            // a cleanup suffix; we never re-throw from finally.
-            if (advised)
+            // A pump failure ends the wait for every tag still outstanding;
+            // tags captured before it keep their values.
+            for (int index = 0; index < waiting.Count; index++)
             {
-                try { UnAdvise(serverHandle, itemHandle); } catch { /* swallow — best effort */ }
+                if (!received[index])
+                {
+                    SnapshotRead failed = waiting[index];
+                    results[failed.Index] = FailedRead(serverHandle, failed.TagAddress, failed.ItemHandle, wasCached: false, exception.Message);
+                }
+            }
+        }
+
+        for (int index = 0; index < waiting.Count; index++)
+        {
+            SnapshotRead snapshot = waiting[index];
+            if (results[snapshot.Index] is not null)
+            {
+                continue;
             }
 
-            if (itemHandle != 0)
+            results[snapshot.Index] = received[index]
+                ? SucceededRead(serverHandle, snapshot.TagAddress, snapshot.ItemHandle, wasCached: false, values[index])
+                : FailedRead(
+                    serverHandle,
+                    snapshot.TagAddress,
+                    snapshot.ItemHandle,
+                    wasCached: false,
+                    $"ReadBulk timed out after {timeout.TotalMilliseconds:F0} ms waiting for first OnDataChange.");
+        }
+    }
+
+    private void TearDownSnapshots(
+        int serverHandle,
+        List<SnapshotRead> snapshots,
+        Action pumpStep)
+    {
+        // Phase 4: snapshot teardown — best-effort, never re-throws. Each
+        // distinct handle is unadvised and removed exactly once, even when
+        // MXAccess handed the same id to several tags in the batch.
+        HashSet<int> unadvised = new();
+        HashSet<int> removed = new();
+        int calls = 0;
+        foreach (SnapshotRead snapshot in snapshots)
+        {
+            if (snapshot.Advised && unadvised.Add(snapshot.ItemHandle))
             {
-                try { RemoveItem(serverHandle, itemHandle); } catch { /* swallow — best effort */ }
+                PumpEvery(ref calls, pumpStep);
+                try { UnAdvise(serverHandle, snapshot.ItemHandle); } catch { /* swallow — best effort */ }
             }
         }
+
+        foreach (SnapshotRead snapshot in snapshots)
+        {
+            if (snapshot.ItemHandle != 0 && removed.Add(snapshot.ItemHandle))
+            {
+                PumpEvery(ref calls, pumpStep);
+                try { RemoveItem(serverHandle, snapshot.ItemHandle); } catch { /* swallow — best effort */ }
+            }
+        }
+    }
+
+    private static void PumpEvery(ref int calls, Action pumpStep)
+    {
+        if (calls > 0 && calls % ReadBulkPumpInterval == 0)
+        {
+            try { pumpStep(); } catch { /* swallow — teardown is best effort */ }
+        }
+
+        calls++;
     }
 
     private bool TryGetCachedReadFor(
@@ -1308,6 +1495,41 @@ public sealed class MxAccessSession : IDisposable
         }
     }
 
+    /// <summary>
+    ///     Mutable per-tag state for one ReadBulk snapshot lifecycle. A plain
+    ///     class with get/set properties (no records or init accessors) so it
+    ///     compiles under the worker's net48 target.
+    /// </summary>
+    private sealed class SnapshotRead
+    {
+        public SnapshotRead(int index, string tagAddress)
+        {
+            Index = index;
+            TagAddress = tagAddress;
+        }
+
+        /// <summary>Gets the position of the tag in the request.</summary>
+        public int Index { get; }
+
+        /// <summary>Gets the requested tag address.</summary>
+        public string TagAddress { get; }
+
+        /// <summary>Gets or sets the snapshot item handle, 0 until AddItem succeeds.</summary>
+        public int ItemHandle { get; set; }
+
+        /// <summary>Gets or sets the value-cache version captured before Advise.</summary>
+        public ulong Baseline { get; set; }
+
+        /// <summary>Gets or sets a value indicating whether Advise succeeded.</summary>
+        public bool Advised { get; set; }
+
+        /// <summary>
+        ///     Gets or sets the earlier snapshot in the same batch whose
+        ///     subscription this one shares (same item handle), or null.
+        /// </summary>
+        public SnapshotRead? SharedWith { get; set; }
+    }
+
     private void ThrowIfDisposed()
     {
         if (disposed)

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

@@ -42,7 +42,15 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
     /// <summary>Default alarm poll cadence when the environment says nothing usable.</summary>
     internal static readonly TimeSpan DefaultAlarmPollInterval = TimeSpan.FromMilliseconds(500);
 
-    private readonly TimeSpan alarmPollInterval = ResolveAlarmPollInterval();
+    /// <summary>
+    ///     Consecutive alarm-poll failures that fault the session. Matches the
+    ///     default of <see cref="FailoverSettings.Threshold"/>
+    ///     (<c>AlarmFailoverConfig.consecutive_failure_threshold</c>, default 3):
+    ///     one transient COM hiccup in <c>PollOnce</c> is logged and retried on
+    ///     the next tick, and only a run of failures with no success in between
+    ///     is treated as a dead subscription.
+    /// </summary>
+    internal const int DefaultAlarmPollFailureThreshold = 3;
 
     private readonly IMxAccessComObjectFactory factory;
     private readonly IMxAccessEventSink eventSink;
@@ -221,6 +229,28 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
     /// </summary>
     internal TimeSpan WriteCompletionTimeout { get; set; } = ResolveWriteCompletionTimeout();
 
+    /// <summary>
+    ///     Cadence the alarm poll loop waits between polls. Resolved from the
+    ///     launcher environment; internal-settable as a test seam (set before
+    ///     <see cref="StartAsync(string, int, CancellationToken)"/>).
+    /// </summary>
+    internal TimeSpan AlarmPollInterval { get; set; } = ResolveAlarmPollInterval();
+
+    /// <summary>
+    ///     Consecutive alarm-poll failures that fault the session; clamped to a
+    ///     minimum of 1. Internal-settable as a test seam.
+    /// </summary>
+    internal int AlarmPollFailureThreshold { get; set; } = DefaultAlarmPollFailureThreshold;
+
+    /// <summary>
+    ///     Sink for the transient alarm-poll failure warning. Defaults to
+    ///     <see cref="Console.Error"/>, the stream <c>WorkerConsoleLogger</c>
+    ///     writes to. The line carries the exception type, HRESULT, and
+    ///     failure counts only — never the exception message, which can echo
+    ///     alarm or tag names. Tests substitute a collector.
+    /// </summary>
+    internal Action<string>? AlarmPollWarningSink { get; set; }
+
     /// <summary>
     ///     Resolves the write-completion wait from the launcher-provided
     ///     environment variable; a missing or invalid value falls back to
@@ -288,6 +318,13 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
     {
         staRuntime.Start();
 
+        // WRK-02: a post-start STA loop death has no other observer. Record it
+        // on the event queue so the IPC drain loop sends a WorkerFault to the
+        // gateway and ends the pipe session (and with it the worker process),
+        // instead of leaving a live process whose commands queue forever.
+        staRuntime.RegisterLoopFaultHandler(
+            exception => eventQueue.RecordFault(CreateStaLoopFault(exception)));
+
         WorkerReady ready = await staRuntime.InvokeAsync(
             () =>
             {
@@ -345,13 +382,16 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
         IAlarmCommandHandler handler,
         CancellationToken cancellationToken)
     {
+        TimeSpan pollInterval = AlarmPollInterval;
+        int failureThreshold = AlarmPollFailureThreshold < 1 ? 1 : AlarmPollFailureThreshold;
         return Task.Run(async () =>
         {
+            int consecutiveFailures = 0;
             while (!cancellationToken.IsCancellationRequested)
             {
                 try
                 {
-                    await Task.Delay(alarmPollInterval, cancellationToken).ConfigureAwait(false);
+                    await Task.Delay(pollInterval, cancellationToken).ConfigureAwait(false);
                 }
                 catch (OperationCanceledException)
                 {
@@ -384,6 +424,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
                             }
                         },
                         cancellationToken).ConfigureAwait(false);
+                    consecutiveFailures = 0;
                 }
                 catch (OperationCanceledException)
                 {
@@ -413,19 +454,66 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
                     // A real alarm-poll failure (COMException from
                     // GetXmlCurrentAlarms2, malformed-XML parse failure, an
                     // STA-affinity InvalidOperationException from
-                    // EnsureOnAlarmConsumerThread, etc.). Record it as a
-                    // fault on the event queue so a broken alarm subscription
-                    // — or an affinity-invariant regression — becomes
+                    // EnsureOnAlarmConsumerThread, etc.). A single failure is
+                    // logged and retried on the next tick: one transient COM
+                    // hiccup must not tear down the whole session (every data
+                    // subscription with it). Only a run of
+                    // AlarmPollFailureThreshold failures with no success in
+                    // between is recorded as a fault on the event queue, so a
+                    // broken alarm subscription — or an affinity-invariant
+                    // regression, which fails every tick — still becomes
                     // observable on the IPC fault path instead of silently
                     // faulting this never-awaited task. The loop then stops —
                     // the subscription is dead.
-                    eventQueue.RecordFault(CreateAlarmPollFault(exception));
-                    return;
+                    if (staRuntime.LoopFault is not null)
+                    {
+                        // The STA loop itself died; its handler already recorded
+                        // the StaHung fault. This failure is just the fast-fail
+                        // InvokeAsync returns afterwards — recording it would add
+                        // a second, mislabelled alarm-poll fault.
+                        return;
+                    }
+
+                    consecutiveFailures++;
+                    if (consecutiveFailures >= failureThreshold)
+                    {
+                        eventQueue.RecordFault(CreateAlarmPollFault(exception));
+                        return;
+                    }
+
+                    WriteAlarmPollWarning(exception, consecutiveFailures, failureThreshold);
                 }
             }
         }, CancellationToken.None);
     }
 
+    private void WriteAlarmPollWarning(Exception exception, int consecutiveFailures, int failureThreshold)
+    {
+        // Same "level=<L> event=<E> k=v" line shape as WorkerConsoleLogger and
+        // the wnwrap truncation warning. Type and HRESULT only — the exception
+        // message is deliberately left out (it can carry alarm or tag names).
+        string hresult = exception is System.Runtime.InteropServices.COMException comException
+            ? "0x" + comException.HResult.ToString("X8", System.Globalization.CultureInfo.InvariantCulture)
+            : "none";
+        string message = string.Format(
+            System.Globalization.CultureInfo.InvariantCulture,
+            "level=Warning event=AlarmPollFailed exceptionType={0} hresult={1} consecutiveFailures={2} faultThreshold={3}",
+            exception.GetType().FullName,
+            hresult,
+            consecutiveFailures,
+            failureThreshold);
+
+        try
+        {
+            Action<string> sink = AlarmPollWarningSink ?? Console.Error.WriteLine;
+            sink(message);
+        }
+        catch
+        {
+            // Logging must never end the poll loop.
+        }
+    }
+
     private void EnsureOnAlarmConsumerThread()
     {
         AssertOnAlarmConsumerThread(alarmConsumerThreadId, Environment.CurrentManagedThreadId);
@@ -458,6 +546,41 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
         }
     }
 
+    /// <summary>
+    ///     Builds the fault reported when the STA loop dies after start. Uses
+    ///     <see cref="WorkerFaultCategory.StaHung"/>: the gateway already treats
+    ///     that category as "this worker's apartment can no longer run COM
+    ///     work", which is exactly the state a dead loop leaves behind.
+    /// </summary>
+    internal static WorkerFault CreateStaLoopFault(Exception exception)
+    {
+        // Type and HRESULT only, never exception.Message — the same type-only
+        // rule every other worker failure line follows.
+        string message = string.Format(
+            System.Globalization.CultureInfo.InvariantCulture,
+            "Worker STA loop terminated unexpectedly: exceptionType={0} hresult=0x{1:X8}",
+            exception.GetType().FullName,
+            exception.HResult);
+        WorkerFault fault = new()
+        {
+            Category = WorkerFaultCategory.StaHung,
+            ExceptionType = exception.GetType().FullName ?? string.Empty,
+            DiagnosticMessage = message,
+            ProtocolStatus = new ProtocolStatus
+            {
+                Code = ProtocolStatusCode.WorkerUnavailable,
+                Message = message,
+            },
+        };
+
+        if (exception is System.Runtime.InteropServices.COMException comException)
+        {
+            fault.Hresult = comException.HResult;
+        }
+
+        return fault;
+    }
+
     private static WorkerFault CreateAlarmPollFault(Exception exception)
     {
         string message =

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

@@ -162,6 +162,46 @@ public sealed class MxAccessValueCache
         }
     }
 
+    /// <summary>
+    ///     Removes every cache slot recorded under a server handle. The session
+    ///     calls this when a server handle is unregistered: MXAccess drops all of
+    ///     that server's items with it, so the per-item <see cref="Remove"/>
+    ///     never runs for them and their entries would otherwise outlive the
+    ///     handle (and could be served to a later server that reuses the id).
+    /// </summary>
+    /// <param name="serverHandle">MXAccess server handle that was unregistered.</param>
+    public void RemoveServer(int serverHandle)
+    {
+        lock (syncRoot)
+        {
+            List<long> stale = new();
+            foreach (long key in entries.Keys)
+            {
+                if ((int)(key >> 32) == serverHandle)
+                {
+                    stale.Add(key);
+                }
+            }
+
+            foreach (long key in stale)
+            {
+                entries.Remove(key);
+            }
+        }
+    }
+
+    /// <summary>Gets the number of handle pairs with a cached value (test seam).</summary>
+    internal int Count
+    {
+        get
+        {
+            lock (syncRoot)
+            {
+                return entries.Count;
+            }
+        }
+    }
+
     /// <summary>
     ///     Waits until the cache entry's version exceeds <paramref name="sinceVersion"/>
     ///     or the deadline elapses, calling <paramref name="pumpStep"/> on every
@@ -231,6 +271,117 @@ public sealed class MxAccessValueCache
         }
     }
 
+    /// <summary>
+    ///     Multi-handle form of <see cref="TryWaitForUpdate"/>: waits until every
+    ///     handle in <paramref name="itemHandles"/> has an entry newer than its
+    ///     matching <paramref name="sinceVersions"/> slot, or the single shared
+    ///     deadline elapses. Each handle is captured the first time an update is
+    ///     observed for it (its first OnDataChange since the baseline), exactly
+    ///     as a dedicated per-handle wait would have captured it, so a bulk
+    ///     snapshot read waits at most one timeout in total rather than one per
+    ///     tag.
+    /// </summary>
+    /// <remarks>
+    ///     Pumps unconditionally at the top of every iteration and blocks the
+    ///     idle part of each iteration in
+    ///     <see cref="StaWaitHelper.WaitForSignalOrMessages"/> — the same pump
+    ///     discipline as <see cref="TryWaitForUpdate"/>, for the same reason.
+    /// </remarks>
+    /// <param name="serverHandle">MXAccess server handle shared by every item.</param>
+    /// <param name="itemHandles">Item handles to wait on.</param>
+    /// <param name="sinceVersions">Per-handle version baselines captured before advising.</param>
+    /// <param name="deadlineUtc">Absolute UTC deadline shared by every handle.</param>
+    /// <param name="pumpStep">Action that pumps any pending Windows messages.</param>
+    /// <param name="received">Output: <see langword="true"/> at index i when handle i received an update before the deadline.</param>
+    /// <param name="values">Output: the captured value at index i when <paramref name="received"/>[i] is set.</param>
+    /// <param name="pollIntervalMs">Upper bound on one idle wait slice.</param>
+    /// <returns>The number of handles that received an update before the deadline.</returns>
+    public int TryWaitForUpdates(
+        int serverHandle,
+        IReadOnlyList<int> itemHandles,
+        IReadOnlyList<ulong> sinceVersions,
+        DateTime deadlineUtc,
+        Action pumpStep,
+        bool[] received,
+        CachedValue[] values,
+        int pollIntervalMs = 5)
+    {
+        if (itemHandles is null)
+        {
+            throw new ArgumentNullException(nameof(itemHandles));
+        }
+
+        if (sinceVersions is null)
+        {
+            throw new ArgumentNullException(nameof(sinceVersions));
+        }
+
+        if (pumpStep is null)
+        {
+            throw new ArgumentNullException(nameof(pumpStep));
+        }
+
+        if (received is null)
+        {
+            throw new ArgumentNullException(nameof(received));
+        }
+
+        if (values is null)
+        {
+            throw new ArgumentNullException(nameof(values));
+        }
+
+        int count = itemHandles.Count;
+        if (sinceVersions.Count != count || received.Length < count || values.Length < count)
+        {
+            throw new ArgumentException("Handle, baseline, and output arrays must have matching lengths.");
+        }
+
+        int receivedCount = 0;
+        for (int index = 0; index < count; index++)
+        {
+            if (received[index])
+            {
+                receivedCount++;
+            }
+        }
+
+        while (true)
+        {
+            pumpStep();
+
+            for (int index = 0; index < count; index++)
+            {
+                if (received[index])
+                {
+                    continue;
+                }
+
+                if (TryGet(serverHandle, itemHandles[index], out CachedValue value) && value.Version > sinceVersions[index])
+                {
+                    received[index] = true;
+                    values[index] = value;
+                    receivedCount++;
+                }
+            }
+
+            if (receivedCount >= count)
+            {
+                return receivedCount;
+            }
+
+            TimeSpan remaining = deadlineUtc - DateTime.UtcNow;
+            if (remaining <= TimeSpan.Zero)
+            {
+                return receivedCount;
+            }
+
+            StaWaitHelper.WaitForSignalOrMessages(
+                updateSignal,
+                StaWaitHelper.ClampWaitMilliseconds(remaining, pollIntervalMs));
+        }
+    }
+
     /// <summary>Returns the current version for a handle pair, or 0 if no entry exists.</summary>
     /// <param name="serverHandle">MXAccess server handle.</param>
     /// <param name="itemHandle">MXAccess item handle.</param>

src/ZB.MOM.WW.MxGateway.Worker/Sta/IStaMessagePump.cs

@@ -0,0 +1,22 @@
+using System;
+using System.Threading;
+
+namespace ZB.MOM.WW.MxGateway.Worker.Sta;
+
+/// <summary>
+///     The two Win32 pump operations <see cref="StaRuntime"/> drives on its
+///     STA thread. <see cref="StaMessagePump"/> is the only production
+///     implementation; the seam exists so Worker.Tests can observe pump
+///     interleaving and inject a pump failure without a real message queue.
+/// </summary>
+internal interface IStaMessagePump
+{
+    /// <summary>Waits for a command wake event or Windows messages.</summary>
+    /// <param name="commandWakeEvent">Event signalled when command work is available.</param>
+    /// <param name="timeout">Maximum time to wait.</param>
+    void WaitForWorkOrMessages(WaitHandle commandWakeEvent, TimeSpan timeout);
+
+    /// <summary>Pumps and dispatches all pending Windows messages.</summary>
+    /// <returns>The number of messages dispatched.</returns>
+    int PumpPendingMessages();
+}

src/ZB.MOM.WW.MxGateway.Worker/Sta/StaMessagePump.cs

@@ -6,7 +6,7 @@ using Microsoft.Win32.SafeHandles;
 namespace ZB.MOM.WW.MxGateway.Worker.Sta;
 
 /// <summary>Pumps Windows messages on the STA thread to allow MXAccess COM events to deliver.</summary>
-public sealed class StaMessagePump
+public sealed class StaMessagePump : IStaMessagePump
 {
     private const uint Infinite = 0xFFFFFFFF;
     private const uint PmRemove = 0x0001;

src/ZB.MOM.WW.MxGateway.Worker/Sta/StaRuntime.cs

@@ -8,7 +8,7 @@ namespace ZB.MOM.WW.MxGateway.Worker.Sta;
 public sealed class StaRuntime : IDisposable
 {
     private readonly IStaComApartmentInitializer comApartmentInitializer;
-    private readonly StaMessagePump messagePump;
+    private readonly IStaMessagePump messagePump;
     private readonly ConcurrentQueue<IStaWorkItem> commandQueue = new();
     private readonly AutoResetEvent commandWakeEvent = new(false);
     private readonly ManualResetEventSlim startedEvent = new(false);
@@ -20,6 +20,8 @@ public sealed class StaRuntime : IDisposable
     private bool startRequested;
     private bool shutdownRequested;
     private Exception? startupException;
+    private Exception? loopFault;
+    private Action<Exception>? loopFaultHandler;
     private long lastActivityUtcTicks;
     private bool comInitialized;
 
@@ -41,6 +43,21 @@ public sealed class StaRuntime : IDisposable
         IStaComApartmentInitializer comApartmentInitializer,
         StaMessagePump messagePump,
         TimeSpan idlePumpInterval)
+        : this(comApartmentInitializer, (IStaMessagePump)messagePump, idlePumpInterval)
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of <see cref="StaRuntime"/> over any
+    /// <see cref="IStaMessagePump"/> (test seam).
+    /// </summary>
+    /// <param name="comApartmentInitializer">COM apartment initializer.</param>
+    /// <param name="messagePump">Message pump for the STA thread.</param>
+    /// <param name="idlePumpInterval">Interval for idle message pump waits.</param>
+    internal StaRuntime(
+        IStaComApartmentInitializer comApartmentInitializer,
+        IStaMessagePump messagePump,
+        TimeSpan idlePumpInterval)
     {
         this.comApartmentInitializer = comApartmentInitializer
             ?? throw new ArgumentNullException(nameof(comApartmentInitializer));
@@ -79,6 +96,59 @@ public sealed class StaRuntime : IDisposable
     /// </summary>
     public bool IsRunning => startedEvent.IsSet && !stoppedEvent.IsSet;
 
+    /// <summary>
+    /// Gets the exception that terminated the STA loop after a successful
+    /// <see cref="Start"/>, or <see langword="null"/> while the loop is healthy
+    /// (or stopped by a requested shutdown). Startup failures are not reported
+    /// here — <see cref="Start"/> throws them.
+    /// </summary>
+    public Exception? LoopFault
+    {
+        get
+        {
+            lock (gate)
+            {
+                return loopFault;
+            }
+        }
+    }
+
+    /// <summary>
+    /// Registers the callback invoked once, on the dying STA thread, when the
+    /// STA loop terminates with an exception after a successful
+    /// <see cref="Start"/>. When the loop has already died the callback runs
+    /// immediately on the calling thread, so a fault that lands between
+    /// <see cref="Start"/> and registration is never lost. Only one handler is
+    /// kept; a later registration replaces an earlier one.
+    /// </summary>
+    /// <remarks>
+    /// Without this, a post-start loop death (a failed
+    /// <c>MsgWaitForMultipleObjectsEx</c>, or an exception escaping a dispatched
+    /// window message) left the worker process alive with no STA: every later
+    /// command queued forever and the gateway only learned of it through the
+    /// heartbeat watchdog, if at all.
+    /// </remarks>
+    /// <param name="handler">Callback receiving the terminating exception.</param>
+    public void RegisterLoopFaultHandler(Action<Exception> handler)
+    {
+        if (handler is null)
+        {
+            throw new ArgumentNullException(nameof(handler));
+        }
+
+        Exception? alreadyFaulted;
+        lock (gate)
+        {
+            loopFaultHandler = handler;
+            alreadyFaulted = shutdownRequested ? null : loopFault;
+        }
+
+        if (alreadyFaulted is not null)
+        {
+            handler(alreadyFaulted);
+        }
+    }
+
     /// <summary>
     ///     Pumps any pending Windows messages on the calling thread and refreshes
     ///     the STA activity timestamp. Intended for commands that synchronously
@@ -176,6 +246,14 @@ public sealed class StaRuntime : IDisposable
 
         lock (gate)
         {
+            if (loopFault is not null)
+            {
+                // The loop is gone; nothing would ever dequeue this item.
+                return Task.FromException<T>(new InvalidOperationException(
+                    "The worker STA loop terminated unexpectedly.",
+                    loopFault));
+            }
+
             if (shutdownRequested)
             {
                 return Task.FromException<T>(new StaRuntimeShutdownException());
@@ -244,12 +322,14 @@ public sealed class StaRuntime : IDisposable
 
     private void ThreadMain()
     {
+        bool started = false;
         try
         {
             StaThreadId = Thread.CurrentThread.ManagedThreadId;
             comApartmentInitializer.Initialize();
             comInitialized = true;
             MarkActivity();
+            started = true;
             startedEvent.Set();
 
             while (!IsShutdownRequested())
@@ -264,8 +344,44 @@ public sealed class StaRuntime : IDisposable
         }
         catch (Exception exception)
         {
-            startupException = exception;
-            startedEvent.Set();
+            if (!started)
+            {
+                // Start() is blocked on startedEvent and rethrows this.
+                startupException = exception;
+                startedEvent.Set();
+            }
+            else
+            {
+                // Post-start loop death. Start() returned long ago, so nothing
+                // would ever read startupException: record the fault where
+                // InvokeAsync and the registered handler see it. Set under the
+                // gate so every item enqueued before it is still in the queue
+                // for the finally block's CancelQueuedCommands, and every
+                // InvokeAsync after it fails fast instead of queueing forever.
+                Action<Exception>? handler;
+                lock (gate)
+                {
+                    loopFault = exception;
+
+                    // A loop that throws while a requested shutdown is already
+                    // in progress is being torn down on purpose; reporting it
+                    // as a worker fault would turn a clean stop into a fault.
+                    handler = shutdownRequested ? null : loopFaultHandler;
+                }
+
+                if (handler is not null)
+                {
+                    try
+                    {
+                        handler(exception);
+                    }
+                    catch
+                    {
+                        // The handler reports the fault; a failure inside it
+                        // must not skip the apartment cleanup below.
+                    }
+                }
+            }
         }
         finally
         {
@@ -292,6 +408,17 @@ public sealed class StaRuntime : IDisposable
             MarkActivity();
             workItem.Execute();
             MarkActivity();
+
+            // Pump between work items, not only once the whole queue is
+            // drained: a burst of queued commands (bulk callers, the alarm
+            // poll, a reconnecting client replaying subscriptions) would
+            // otherwise hold every inbound COM callback — OnDataChange,
+            // OnWriteComplete — behind the entire backlog. PeekMessage with an
+            // empty queue is cheap, so the per-item cost is negligible.
+            if (!commandQueue.IsEmpty)
+            {
+                messagePump.PumpPendingMessages();
+            }
         }
     }
 

src/ZB.MOM.WW.MxGateway.Worker/WorkerApplication.cs

@@ -14,10 +14,61 @@ public static class WorkerApplication
     /// <returns>The process exit code.</returns>
     public static int Run(string[] args)
     {
+        WorkerConsoleLogger logger = new(Console.Error);
+
+        // WRK-09: installed only on the production entry point (not the
+        // injectable overloads tests call) because the hook is process-global.
+        AppDomain.CurrentDomain.UnhandledException += (_, eventArgs) =>
+            HandleUnhandledException(
+                logger,
+                eventArgs.ExceptionObject,
+                eventArgs.IsTerminating,
+                Environment.Exit);
+
         return Run(
             args,
             new EnvironmentVariableWorkerEnvironment(),
-            new WorkerConsoleLogger(Console.Error));
+            logger);
+    }
+
+    /// <summary>
+    ///     Body of the worker's <c>AppDomain.CurrentDomain.UnhandledException</c>
+    ///     hook. Logs one <c>WorkerUnhandledException</c> line and exits the
+    ///     process with <see cref="WorkerExitCode.UnhandledException"/>, so the
+    ///     gateway's launcher sees a defined worker exit code rather than the
+    ///     CLR's crash code.
+    /// </summary>
+    /// <remarks>
+    ///     Logs the exception type only — never its message or stack, which
+    ///     can carry tag values or credentials passed through a command — the
+    ///     same rule every other worker failure line follows. A logging
+    ///     failure is swallowed so the exit always happens.
+    /// </remarks>
+    /// <param name="logger">Worker logger.</param>
+    /// <param name="exceptionObject">The unhandled exception object (not necessarily an <see cref="Exception"/>).</param>
+    /// <param name="isTerminating">Whether the CLR is terminating the process.</param>
+    /// <param name="exit">Process-exit action; <see cref="Environment.Exit"/> in production.</param>
+    internal static void HandleUnhandledException(
+        IWorkerLogger logger,
+        object? exceptionObject,
+        bool isTerminating,
+        Action<int> exit)
+    {
+        try
+        {
+            logger.Error("WorkerUnhandledException", new Dictionary<string, object?>
+            {
+                ["exit_code"] = WorkerExitCode.UnhandledException,
+                ["exception_type"] = exceptionObject?.GetType().FullName ?? "null",
+                ["is_terminating"] = isTerminating,
+            });
+        }
+        catch
+        {
+            // Never let a logging failure skip the defined exit.
+        }
+
+        exit((int)WorkerExitCode.UnhandledException);
     }
 
     /// <summary>Initializes and runs the worker with custom environment and logging.</summary>