chore(plans): tasks 1-12 complete through review chains

The two-class writer already got control bytes out ahead of a queued event
backlog, but a frame counts as delivered only once flushed, and the drain
deferred its single FlushAsync — and every TrySetResult — to the end of the
pass. A heartbeat, command reply, fault, or shutdown ack was therefore written
first and completed last, behind up to a full 128-frame event batch.

The drain now records each frame's priority class on PendingFrame and flushes
at every control-to-event boundary, completing and clearing the written set
there. Cost stays bounded: a pure-event pass still pays exactly one flush, a
run of control frames still pays one for the run, and only a pass that mixes
both classes pays a second — never one flush per control frame, the
syscall-per-heartbeat cost WRK-12 removed.

A boundary flush that itself fails is a new failure window and is handled like
the end-of-pass flush failure, additionally failing the event frame the drain
had already claimed off its queue and every frame still queued. Frames a
boundary flush completed leave the written set, so a later failure in the same
pass can no longer reach back and fail an already-delivered control frame.

The awaited task of a caller that lost the write-lock race is still bounded by
the winning drainer's pass — that enqueue-then-contend parking is unchanged and
now documented on WriteAsync and in docs/WorkerFrameProtocol.md.

docs/GatewayDashboardDesign.md

@@ -98,6 +98,7 @@ ZB.MOM.WW.MxGateway.Server
         StatusBadge.razor
         FaultList.razor
     DashboardSnapshotService.cs
+    DashboardSnapshotFeed.cs
     DashboardAuthorizationHandler.cs
     DashboardAuthenticator.cs
     DashboardApiKeyAuthorization.cs
@@ -110,9 +111,15 @@ ZB.MOM.WW.MxGateway.Server
 ```
 
 The dashboard exposes three named SignalR hubs in addition to Blazor Server's
-internal circuit; pages connect to those hubs from within the circuit via the
-`DashboardHubConnectionFactory` helper. The hubs publish snapshot, alarm, and
-per-session event updates that the pages render in place of polling.
+internal circuit. The hubs are the **remote** surface: they publish snapshot,
+alarm, and per-session event updates to clients outside the gateway process.
+Server-rendered Blazor pages do not use them. A page runs inside this process,
+so it consumes the producing services directly through in-process seams —
+`IDashboardSnapshotFeed`, `IDashboardSessionEventSubscriber`, and
+`IGatewayAlarmService` — instead of opening a loopback WebSocket back into its
+own heap. `DashboardHubConnectionFactory`, the helper a circuit used to open
+those connections, stays registered for out-of-tree consumers, but no in-repo
+page resolves it.
 
 ## Dashboard Data Source
 
@@ -159,7 +166,57 @@ gateway internals.
 
 ## Realtime Updates
 
-Updates flow over three SignalR hubs, all guarded by the
+Realtime data reaches two audiences over two seams:
+
+- **in-process**, for the server-rendered Blazor pages, which run inside the
+  gateway process and read the producing services directly;
+- **SignalR hubs**, for clients outside the process.
+
+Pages originally took the hub path too, which put a loopback WebSocket, a
+hub-token mint, and a serialize/deserialize round trip between a Blazor component
+and an object already in its own heap. The in-process seams remove that hop. The
+hubs stay for the audience that genuinely needs a wire.
+
+### In-process page feeds
+
+| Page | Seam | Producer |
+|---|---|---|
+| every page deriving from `DashboardPageBase` | `IDashboardSnapshotFeed.WatchAsync` | `DashboardSnapshotFeed` (singleton) multicasting one `IDashboardSnapshotService.WatchSnapshotsAsync` enumeration |
+| `SessionDetailsPage` | `IDashboardSessionEventSubscriber.Subscribe(sessionId)` | `DashboardEventBroadcaster` — the same singleton the session mirror publishes to, registered behind both interfaces |
+| `AlarmsPage` | `IGatewayAlarmService.StreamAsync` | the central alarm monitor, **provider status only**; the alarm rows still come from the 3 s `QueryAlarmsAsync` poll |
+
+The snapshot feed multicasts rather than handing each page its own enumeration:
+`WatchSnapshotsAsync` is not multicast on its own — each enumeration owns a timer
+and builds its own snapshot per tick — so a subscription per page would multiply
+the snapshot cost by the number of open pages. Each subscriber reads through a
+capacity-1 drop-oldest channel, so a circuit that renders slowly skips snapshots
+instead of buffering without bound or stalling the pump.
+
+`DashboardPageBase` seeds `Snapshot` synchronously from
+`IDashboardSnapshotService.GetSnapshot()` in `OnInitializedAsync` so the first
+render is non-empty, then calls `InvokeAsync(StateHasChanged)` for every snapshot
+the feed yields. On dispose it cancels the watch and waits at most **5 seconds**
+for the loop to drain, logging a warning on timeout. The bound is deliberate: the
+loop marshals renders through the renderer's dispatcher and disposal can run on
+that same dispatcher, so an unconditional wait would hang on a wedged dispatcher.
+The accepted cost is that an abandoned loop still holds its feed subscription — the
+feed's idle gate stays open until it unwinds — and the warning is the operator's
+only signal that a circuit teardown wedged.
+
+`SessionDetailsPage` subscribes for the current session id and renders the most
+recent N events (default 50) in a "Recent events" table. Its pump drains everything
+queued and renders once per batch rather than once per event, and it re-checks
+**inside the renderer dispatch** — where the subscription field is written, making
+the check an unsynchronized read of dispatcher-owned state — that the batch's
+subscription is still the live one. A batch read before a session switch would
+otherwise render the previous session's events under the new session's heading.
+Detaching cancels the pump, disposes the subscription (which releases the viewer
+registration and completes the channel, so the pump has an exit even if
+cancellation is missed), then drains under its own timeout.
+
+### SignalR hubs (remote clients)
+
+Updates for out-of-process clients flow over three SignalR hubs, all guarded by the
 `MxGateway.Dashboard.HubClients` policy (cookie OR `MxGateway.Dashboard.HubToken`
 bearer). Each hub class is `[Authorize(Policy = HubClientsPolicy)]`.
 
@@ -167,46 +224,68 @@ bearer). Each hub class is `[Authorize(Policy = HubClientsPolicy)]`.
 |---|---|---|---|---|
 | `DashboardSnapshotHub` | `/hubs/snapshot` | `DashboardSnapshotPublisher` (BackgroundService consuming `IDashboardSnapshotService.WatchSnapshotsAsync`) | `DashboardSnapshot` | Sent to all connected clients on every snapshot tick, but only while at least one client is connected (see "Idle gating" below); new connections receive the current snapshot synchronously in `OnConnectedAsync`. |
 | `AlarmsHub` | `/hubs/alarms` | `AlarmsHubPublisher` (BackgroundService consuming `IGatewayAlarmService.StreamAsync(filter: null)`) | `AlarmFeedMessage` (`active_alarm` / `snapshot_complete` / `transition`) | Connected clients auto-join `__alarms__`; all clients receive every message. Publisher auto-reconnects every 5s on stream faults. |
-| `EventsHub` | `/hubs/events` | `DashboardEventBroadcaster` invoked by each session's internal dashboard-mirror subscriber on its `SessionEventDistributor` (registered when the session becomes Ready) | `MxEvent` | Clients call `SubscribeSession(sessionId)` to join `session:{id}`, which also registers them in `EventsHubViewerRegistry` — the mirror is gated on that registry (see "Mirror gating" below). The dashboard is a first-class distributor subscriber, so it receives the session's events whether or not a gRPC client is streaming. It sees RAW session events — not the per-gRPC-subscriber `AfterWorkerSequence` filtering that `EventStreamService` applies at its own boundary — because the dashboard is a separate LDAP-authenticated monitoring view meant to show the session's full event activity. Tag values are stripped from the mirrored `MxEvent` copy by `DashboardEventBroadcaster` when `Dashboard:ShowTagValues` is false (the default) — event metadata (tag reference, quality, status, timestamps) still renders, but the value fields are blanked, so no value leaks through this seam. The per-session hub ACL that would scope a Viewer to specific sessions is still outstanding (SEC-25 / remediation roadmap item 12); the value redaction is the near-term hardening that closes the value-leak seam independently of that ACL. |
+| `EventsHub` | `/hubs/events` | `DashboardEventBroadcaster` invoked by each session's internal dashboard-mirror subscriber on its `SessionEventDistributor` (registered when the session becomes Ready) | `MxEvent` | Clients call `SubscribeSession(sessionId)` to join `session:{id}`, which also registers them in `EventsHubViewerRegistry` — the mirror is gated on that registry, which counts hub and in-process viewers alike (see "Mirror gating" below). The dashboard is a first-class distributor subscriber, so it receives the session's events whether or not a gRPC client is streaming. It sees RAW session events — not the per-gRPC-subscriber `AfterWorkerSequence` filtering that `EventStreamService` applies at its own boundary — because the dashboard is a separate LDAP-authenticated monitoring view meant to show the session's full event activity. Tag values are stripped from the mirrored `MxEvent` copy by `DashboardEventBroadcaster` when `Dashboard:ShowTagValues` is false (the default) — event metadata (tag reference, quality, status, timestamps) still renders, but the value fields are blanked, so no value leaks through this seam. The per-session ACL that would scope a Viewer to specific sessions is still outstanding for this seam and the in-process one alike (SEC-25 / remediation roadmap item 12); the value redaction is the near-term hardening that closes the value-leak seam independently of that ACL. |
 
-`DashboardPageBase` opens a `DashboardSnapshotHub` connection via the connection
-factory in `OnInitializedAsync`, seeds `Snapshot` synchronously from
-`IDashboardSnapshotService.GetSnapshot()` so the first render is non-empty, and
-calls `InvokeAsync(StateHasChanged)` on every `SnapshotUpdated` push. SignalR's
-`WithAutomaticReconnect` handles transient disconnects.
+### Default cadences
 
-`SessionDetailsPage` additionally opens an `EventsHub` connection for the
-current session id and renders the most recent N events (default 50) in a
-"Recent events" table with a live/offline connection pill.
-
-Default cadences:
+Both seams consume the same producing services, so they share these cadences:
 
 - snapshot service produces one snapshot per
   `MxGateway:Dashboard:SnapshotIntervalMilliseconds` (default 1s);
 - alarm publisher emits on each transition observed by the central monitor;
 - event publisher emits per event fanned by the session's `SessionEventDistributor`
-  to its internal dashboard-mirror subscriber (independent of any gRPC `StreamEvents`).
+  to its internal dashboard-mirror subscriber (independent of any gRPC `StreamEvents`);
+- the alarms page's provider-status badge resubscribes one second after its
+  `IGatewayAlarmService.StreamAsync` enumeration ends — the monitor completes a
+  subscriber's stream when it falls behind and again when it restarts, both
+  recoverable by resubscribing — and holds its last value in between. The page's
+  alarm rows are independent of that stream and refresh on the 3 s poll.
 
 ### Idle gating and snapshot cost
 
 A snapshot is not free: each one takes a session-registry snapshot and sorts it,
 copies the metrics dictionaries under the global metrics lock, and projects
 sessions, workers, faults, and the Galaxy summary. Without gating that work ran
-once a second for the life of the process even when no browser was connected.
+once a second for the life of the process even when nothing was watching.
 
-`DashboardSnapshotHub` counts live connections into the singleton
+Gating is two-tier, because the two seams have independent audiences and each must
+be able to reach zero on its own.
+
+**Hub tier.** `DashboardSnapshotHub` counts live connections into the singleton
 `DashboardSnapshotHubConnectionCounter` (`OnConnectedAsync` / `OnDisconnectedAsync`,
 clamped at zero). `DashboardSnapshotPublisher` reads that count before advancing the
 snapshot enumerator: while it is zero the publisher does not call `MoveNextAsync` at
 all, so the producing iterator stays suspended at its `yield` and builds nothing —
 the gate removes the snapshot *build*, not just the broadcast. The publisher
-re-checks once a second while idle, so the first viewer to connect resumes the tick
-within roughly one snapshot interval. That viewer does not wait for it either:
-`DashboardPageBase` seeds its first render synchronously from
-`IDashboardSnapshotService.GetSnapshot()`, and `OnConnectedAsync` pushes a snapshot
-to the new connection immediately.
+re-checks once a second while idle, so the first client to connect resumes the tick
+within roughly one snapshot interval, and `OnConnectedAsync` pushes the current
+snapshot to that connection immediately. Now that no in-repo page connects to the
+hub, this tier stays idle unless a remote client connects.
 
-Two per-tick costs inside the snapshot itself are bounded independently of the gate:
+**In-process tier.** `DashboardSnapshotFeed` gates its own pump on its subscriber
+list: the first subscriber starts the pump, the last one leaving cancels it and
+awaits it, so a gateway with no page open runs no timer and builds no snapshots on
+this seam either. Successive pumps are chained through the previous pump's task, so
+an unsubscribe immediately followed by a resubscribe restarts a fresh pump without
+ever running two enumerations at once. A page does not wait for the pump's first
+tick — `DashboardPageBase` seeds its first render synchronously from
+`IDashboardSnapshotService.GetSnapshot()`.
+
+That gate is generation-scoped rather than a plain subscriber count. Each pump owns
+a generation, each subscriber is tagged with the generation it joined under, a pump
+ends its generation the instant its source faults or completes — before the possibly
+slow enumerator disposal — and a dying pump only ever detaches its own generation's
+subscribers. Two races motivate the extra state:
+
+- a subscriber arriving mid-teardown must start a fresh generation rather than
+  attach to a pump that is about to detach everybody and leave nobody watching;
+- an unsubscribe must compare its own generation against the live one before it
+  cancels anything. Subscribers of an ending generation linger in the list until
+  that pump's reset runs, so counting the whole list would let them hold the idle
+  gate open, and cancelling on their behalf would stop a live pump that other
+  viewers depend on.
+
+Two per-tick costs inside the snapshot itself are bounded independently of either gate:
 
 - the effective configuration (`EffectiveGatewayConfiguration`) is built once and
   cached. It is a projection of `IOptions<GatewayOptions>`, which the gateway binds
@@ -220,26 +299,43 @@ Two per-tick costs inside the snapshot itself are bounded independently of the g
   retried on the next tick and the previous summaries stay on screen.
 
 Avoid pushing every MXAccess data-change event into a wider broadcast group.
-The current design routes events strictly through `session:{id}` groups; the
-snapshot hub continues to carry aggregate event counters and rates.
+Events are routed strictly per session (`session:{id}` groups on the hub,
+per-session subscriber lists in process); the snapshot seams continue to carry
+aggregate event counters and rates.
 
 ### Mirror gating
 
 Each session's dashboard-mirror subscriber calls
 `DashboardEventBroadcaster.Publish` for every event the session produces,
-independently of whether any browser is watching that session. SignalR does not
-expose group membership, so the broadcaster cannot ask whether `session:{id}` is
-empty. `EventsHubViewerRegistry` (singleton) supplies that answer: `EventsHub`
-mirrors its own `AddToGroup` / `RemoveFromGroup` calls into it, and
-`OnDisconnectedAsync` releases every subscription a dropped connection held — the
-only reliable signal for a browser tab that closes without unsubscribing.
-`Publish` returns immediately when `HasViewers(sessionId)` is false, **before**
-the redaction clone. That matters because redaction is on by default
+independently of whether anything is watching that session. `Publish` returns
+immediately when `EventsHubViewerRegistry.HasViewers(sessionId)` is false,
+**before** the redaction clone. That matters because redaction is on by default
 (`Dashboard:ShowTagValues` false), so the unwatched steady state — nobody on any
 session-details page — previously paid a deep protobuf clone plus a send to an
 empty group for every event of every session. Behaviour for a watched session is
 unchanged.
 
+The registry counts both audiences, which is what lets one gate serve both seams.
+`EventsHub` mirrors its own `AddToGroup` / `RemoveFromGroup` calls into it —
+SignalR does not expose group membership, so the broadcaster cannot ask whether
+`session:{id}` is empty — and `OnDisconnectedAsync` releases every subscription a
+dropped connection held, the only reliable signal for a browser tab that closes
+without unsubscribing. An in-process subscription registers the same way under a
+synthetic `inproc-`-prefixed connection id, which cannot collide with a SignalR
+connection id and makes the origin obvious in a debugger; disposing it removes the
+viewer and releases the synthetic connection, because that id never reconnects and
+nothing else would ever release its per-connection entry. Both paths use the same
+ordering — register before becoming a delivery target, deregister after ceasing to
+be one — so the widest a race window opens is a redaction clone that reaches
+nobody, never a dropped event that was owed to a live viewer.
+
+Redaction happens once per event, not once per audience: `Publish` produces a
+single redacted clone and hands that same instance to the in-process subscribers
+and to the hub group. In-process delivery runs first and synchronously — it cannot
+throw, and it must not be skipped by the guard clause around the hub send — into
+per-subscriber bounded drop-oldest channels, so a page that falls behind loses its
+oldest queued events rather than blocking the session's event pipeline.
+
 The mirror subscriber itself is still registered on the `SessionEventDistributor`
 for the session's whole lifetime; only the per-event work is gated. Starting and
 stopping the mirror lease lazily with the first and last viewer was considered
@@ -369,8 +465,9 @@ panel. The panel shows each subscribed tag's live value, MXAccess data type,
 quality and source timestamp, refreshed every two seconds. The subscription
 panel is the explicit opt-in tag-value surface: it always shows values
 regardless of `Dashboard:ShowTagValues`, which governs the diagnostic
-session/worker views and the per-session `EventsHub` mirror (values are
-redacted from the mirrored events when the flag is false).
+session/worker views and the per-session event mirror — both its hub and
+in-process audiences (values are redacted from the mirrored events when the flag
+is false).
 
 ### Alarms page
 
@@ -380,7 +477,11 @@ defaults to showing unacknowledged `Active` alarms; filters add acknowledged
 alarms and narrow by area, severity range, and a reference/source/description
 text search. Cleared alarms are not retained — the gateway holds no
 alarm-history store, so the page reflects only the live active set. The page is
-read-only; it does not acknowledge alarms. If `MxGateway:Alarms:Enabled` is
+read-only; it does not acknowledge alarms. A provider-status badge tracks the
+central monitor's health from `IGatewayAlarmService.StreamAsync` in process — the
+alarm service is already a multi-subscriber fan-out, so the badge needs no SignalR
+client, no loopback socket, and no hub token — while the alarm rows themselves
+still come from the three-second poll. If `MxGateway:Alarms:Enabled` is
 false the central monitor never starts, and the page says so instead of showing
 an empty list with no explanation.
 
@@ -536,6 +637,14 @@ Three authorization policies are registered:
   cookie OR a `MxGateway.Dashboard.HubToken` bearer (used by WebSocket upgrades
   where the cookie can't be forwarded).
 
+The in-process page feeds carry no authentication of their own, and need none:
+`MapRazorComponents<App>()` applies `RequireAuthorization(ViewerPolicy)` to the
+component endpoints, so a page can only run inside a circuit whose principal is
+already an authorized Viewer. The hub-token flow below therefore covers only the
+remote hub surface. Neither seam scopes a Viewer to particular sessions — SEC-25
+(the per-session ACL) is outstanding for both, and the mirror's value redaction
+remains the near-term mitigation, unchanged by the move in-process.
+
 Two environmental bypasses still apply, both scoped to **read-only** access:
 `MxGateway:Authentication:Mode = Disabled` and `MxGateway:Dashboard:AllowAnonymousLocalhost`
 (default `true`, loopback only) each satisfy a requirement that includes the Viewer
@@ -580,12 +689,15 @@ surface is affected. Never enable in production.
 
 ### Hub bearer flow
 
+This flow serves remote hub clients only; in-process pages are authorized by the
+component endpoint's `ViewerPolicy` and never mint a token.
+
 SignalR connections cannot reuse the `__Host-` cookie when the JS client
 upgrades to WebSocket — the cookie's `SameSite=Strict; Path=/` keeps it from
 being forwarded by the browser's WebSocket layer in some edge cases. The
 dashboard mints short-lived bearer tokens for the connection:
 
-1. The cookie-authenticated Blazor page calls `GET /hubs/token`
+1. The cookie-authenticated client calls `GET /hubs/token`
    (gated by `ViewerPolicy`, cookie-only).
 2. `HubTokenService.Issue(user)` serializes the user's name, NameIdentifier,
    and role claims to JSON, encrypts with the ASP.NET Core data-protection
@@ -605,7 +717,9 @@ dashboard mints short-lived bearer tokens for the connection:
 
 `DashboardHubConnectionFactory` (scoped to the Blazor circuit) wraps the
 HubConnectionBuilder and supplies a fresh token via `AccessTokenProvider` on
-every (re)connect, so the short 5-minute lifetime is transparent to clients.
+every (re)connect, so the short 5-minute lifetime is transparent to whoever uses
+it. It remains registered, but no in-repo page opens a hub connection any more;
+external clients implement the equivalent refresh themselves.
 
 Caveat — logout does not revoke outstanding tokens. Logout clears the dashboard
 cookie, but hub bearer tokens are self-contained, data-protection-encrypted, and
@@ -700,8 +814,14 @@ Integration tests should verify:
 - a user in a Viewer-mapped LDAP group can render every page but cannot
   invoke the Admin-only management actions,
 - a user with no mapped LDAP group cannot sign in at all,
-- live snapshot updates when a fake session changes state are delivered
-  via the `/hubs/snapshot` push, not by polling.
+- live snapshot updates when a fake session changes state reach a page through
+  the in-process `IDashboardSnapshotFeed` and reach a remote client through the
+  `/hubs/snapshot` push — neither by polling;
+- the snapshot feed's idle gate: no subscribers means no pump, the last
+  subscriber of the live generation stops it, and a subscriber arriving
+  mid-teardown gets a fresh generation rather than a dead one;
+- the event mirror's viewer gate counts in-process subscriptions as well as hub
+  connections, and a disposed in-process subscription releases its viewer count.
 
 ## Initial Implementation Slice
 

docs/MxAccessWorkerInstanceDesign.md

@@ -887,6 +887,50 @@ Graceful shutdown sequence:
 If shutdown wedges, the gateway kills the process. The worker should be written
 so process kill does not corrupt other sessions.
 
+### Ending the pipe read (net48)
+
+Step 8 above cannot be done by cancellation. On .NET Framework 4.8
+`NamedPipeClientStream.ReadAsync` accepts a `CancellationToken` and then never
+wires it to the overlapped I/O, so a read parked waiting for gateway bytes stays
+parked no matter what the worker cancels. Closing the handle is the only thing
+that ends it.
+
+`WorkerPipeSession.RunMessageLoopAsync` races one outstanding read against the
+heartbeat and event-drain loops, so every fault exit — an event-drain fault, an
+event too large to frame, a failed heartbeat write — unwinds while that read is
+still pending. The session therefore owns the transport: `RunAsync`'s outermost
+`finally` disposes the stream as its last teardown step and then awaits the read
+that disposal unblocks. Disposal comes last because in the ordinary case every
+frame the session will ever write is already complete by then — the frame writer
+signals a write only after it has been written *and* flushed. It is not last
+because that is guaranteed: the wait on the heartbeat and drain loops is
+budgeted, and a stream write is genuinely uncancellable, so an overrunning write
+can still be in flight against the stream being disposed. Disposal is
+consequently exception-*total*, catching anything the handle close throws and
+logging it, because nothing raised while releasing a handle is more actionable
+than the terminal exception that ended the session, and nothing may displace it.
+
+Observation is unconditional; only the *logging* of it is budgeted.
+`ObserveBackgroundTaskStopAsync` waits `BackgroundTaskStopTimeout` for a task to
+stop and logs what it saw, but when it gives up it hands the task a
+fault-observing continuation before returning. Windows owes no deadline for a
+completion torn off a closed handle, so a bounded await on its own would reopen
+the very orphaning window it was added to close. The same helper — and so the
+same guarantee — covers the abandoned read, the heartbeat loop, and the
+event-drain loop. This matters because the worker installs no
+`TaskScheduler.UnobservedTaskException` handler: an unheld faulted task would
+otherwise surface only at finalization, still holding the reader's reused
+length-prefix buffer and its pooled payload buffer.
+
+Two invariants follow. Nothing may call `WorkerFrameReader.ReadAsync` again once
+a read has been abandoned — a second read would race the first for those buffers
+and could return a pooled buffer twice — which `Debug.Assert`s at both read-issue
+sites guard. And `WorkerPipeClient`'s `using` on the pipe stays as a backstop for
+the paths the session never reaches (a session factory that throws), not as the
+primary owner; disposal is idempotent, so its second `Dispose` is a no-op.
+Graceful shutdown leaves no pending read at all, so the observation step is a
+no-op on that path.
+
 `MxAccessStaSession.ShutdownGracefullyAsync` implements the current cleanup
 path. It first calls `StaCommandDispatcher.RequestShutdown()` so new commands
 are rejected and queued commands that have not started receive

docs/WorkerFrameProtocol.md

@@ -88,7 +88,9 @@ priority order. A caller enqueues its frame into the control or event queue
 under a lock, then contends for a single write lock; whichever caller wins
 drains every frame queued at that moment, control frames first and each class
 in FIFO order, so a command reply, fault, heartbeat, or shutdown
-acknowledgement is never delayed behind a backlog of queued events. Priority
+acknowledgement is never delayed behind a backlog of queued events — neither
+in the bytes written nor in the flush that marks them delivered (see the
+class-boundary flush under flush coalescing below). Priority
 only reorders *which frame writes next* — it does not affect the sequence
 value a frame receives (see below), so a caller cannot infer priority class
 from the wire sequence.
@@ -117,13 +119,36 @@ Two failure shapes are distinguished during a drain pass:
   and every frame still queued, then stops draining entirely so no caller
   waits forever on a stream that will not recover.
 
-Flushes are coalesced across a drained batch: each frame in the batch is
-written to the stream without an individual flush, then one `FlushAsync`
-runs after the whole batch, and only then does every successfully-written
-frame's completion resolve — so a caller's `WriteAsync` still does not
-complete until its bytes are both written *and* flushed, but a batch that
-happened to contain several queued frames pays one flush instead of one per
-frame. Note the ordering this implies at the peer: the frames reach the pipe
+Flushes are coalesced across a *run of same-class frames* inside a drain
+pass: each frame in the run is written to the stream without an individual
+flush, then one `FlushAsync` runs — at the end of the pass, and additionally
+at every control-to-event boundary — and only then does every
+successfully-written frame of that run resolve its completion. A caller's
+`WriteAsync` therefore still does not complete until its bytes are both
+written *and* flushed; what changed is *when* that moment arrives
+for a control frame that a pass writes ahead of queued events. It used to be
+the end of the pass, so a heartbeat, command reply, fault, or shutdown
+acknowledgement was written first but only counted as delivered after up to a
+full event batch had been written and flushed behind it. The boundary flush
+closes the control run out before the events are written, so the priority
+class governs the frame's delivery point and not just its byte order. The
+cost stays bounded: a pure-event pass — the event hot path — still pays
+exactly one flush however many frames drain together, a run of control
+frames still pays one for the whole run (never one per heartbeat, the
+syscall-per-frame cost the coalescing removed), and only a pass that actually
+mixes both classes pays a second.
+
+One consequence of the boundary flush is worth stating: a control frame whose
+run has already been flushed and completed is out of the drain's
+written-but-unflushed set, so a *later* failure in the same pass — a broken
+write, or a failed end-of-pass flush — no longer reaches back and fails it.
+That is the honest outcome: its bytes were flushed, so it was delivered. A
+failure of the boundary flush itself is treated exactly like a failed
+end-of-pass flush, and additionally fails the event frame the drain had
+already claimed off its queue (nothing else would ever complete it) along
+with every frame still queued.
+
+Note the ordering all of this implies at the peer: the frames reach the pipe
 before the flush that follows them, so the gateway can read a whole batch
 while the writer has not yet flushed it. Anything observing the flush itself
 (a test counting flushes, for instance) must wait for the flush, not infer it
@@ -134,12 +159,22 @@ drains them together, so a burst of N events costs one flush rather than N —
 the coalescing the batch machinery was built for now engages on the event hot
 path, not only when independent producers happen to queue behind a blocked
 write. Intra-batch order is preserved (FIFO enqueue under one lock), and a
-concurrently queued control frame is still drained ahead of the batch. A
-per-frame rejection inside a batch (for example one oversized event) surfaces
-from the batch's awaited completions as that frame's
+concurrently queued control frame is still drained — and now flushed and
+completed — ahead of the batch's remaining events, which is why a batch a
+control frame cuts into pays one extra flush while an uninterrupted batch
+still pays exactly one. A per-frame rejection inside a batch (for example one
+oversized event) surfaces from the batch's awaited completions as that frame's
 `WorkerFrameProtocolException`; the remaining completions are still observed
 so none faults unobserved.
 
+The completion is the frame's delivery point, not necessarily the instant its
+caller returns. A caller that loses the race for the write lock only observes
+its own completion after the winning drainer releases the lock, so its return
+remains bounded by that drain pass even though its control frame was flushed
+and completed at the class boundary inside it. The boundary flush is what
+makes the delivery point honest; unparking a lock-race loser from the winner's
+pass would be a separate change to the enqueue-then-contend shape.
+
 Cancellation of a `WriteAsync`/`WriteBatchAsync` call that is still waiting
 for the write lock when its token fires tombstones the queued frame: the
 cancelled caller marks its frame under `_gate`, and the draining lock-holder's

docs/plans/2026-08-15-deferred-remediation.md.tasks.json

@@ -1,18 +1,18 @@
 {
   "planPath": "docs/plans/2026-08-15-deferred-remediation.md",
   "tasks": [
-    { "id": 1, "subject": "Task 1: Windows-safe cleanup in SecretsStorePathGuardTests", "status": "pending" },
-    { "id": 2, "subject": "Task 2: SessionEventDistributor _subscribers to plain Dictionary", "status": "pending" },
-    { "id": 3, "subject": "Task 3: Merge the session event-source pass-through iterator", "status": "pending" },
-    { "id": 4, "subject": "Task 4: EventStreamService direct channel reads in the live loop", "status": "pending" },
-    { "id": 5, "subject": "Task 5: In-process dashboard snapshot feed + page switch", "status": "pending" },
-    { "id": 6, "subject": "Task 6: In-process session event subscription + SessionDetailsPage switch", "status": "pending" },
-    { "id": 7, "subject": "Task 7: AlarmsPage provider-status via IGatewayAlarmService", "status": "pending" },
-    { "id": 8, "subject": "Task 8: Dashboard design-doc update (consolidated)", "status": "pending", "blockedBy": [5, 6, 7] },
-    { "id": 9, "subject": "Task 9: Phase A gate — full gateway suite on macOS", "status": "pending", "blockedBy": [1, 2, 3, 4, 5, 6, 7, 8] },
-    { "id": 10, "subject": "Task 10: Control-frame completion decoupling in WorkerFrameWriter", "status": "pending", "blockedBy": [9] },
-    { "id": 11, "subject": "Task 11: Worker pipe-read teardown — dispose-to-unblock and observe", "status": "pending", "blockedBy": [9] },
-    { "id": 12, "subject": "Task 12: Value-cache clone removal per the aliasing audit", "status": "pending", "blockedBy": [9] },
+    { "id": 1, "subject": "Task 1: Windows-safe cleanup in SecretsStorePathGuardTests", "status": "completed" },
+    { "id": 2, "subject": "Task 2: SessionEventDistributor _subscribers to plain Dictionary", "status": "completed" },
+    { "id": 3, "subject": "Task 3: Merge the session event-source pass-through iterator", "status": "completed" },
+    { "id": 4, "subject": "Task 4: EventStreamService direct channel reads in the live loop", "status": "completed" },
+    { "id": 5, "subject": "Task 5: In-process dashboard snapshot feed + page switch", "status": "completed" },
+    { "id": 6, "subject": "Task 6: In-process session event subscription + SessionDetailsPage switch", "status": "completed" },
+    { "id": 7, "subject": "Task 7: AlarmsPage provider-status via IGatewayAlarmService", "status": "completed" },
+    { "id": 8, "subject": "Task 8: Dashboard design-doc update (consolidated)", "status": "completed", "blockedBy": [5, 6, 7] },
+    { "id": 9, "subject": "Task 9: Phase A gate — full gateway suite on macOS", "status": "completed", "blockedBy": [1, 2, 3, 4, 5, 6, 7, 8] },
+    { "id": 10, "subject": "Task 10: Control-frame completion decoupling in WorkerFrameWriter", "status": "completed", "blockedBy": [9] },
+    { "id": 11, "subject": "Task 11: Worker pipe-read teardown — dispose-to-unblock and observe", "status": "completed", "blockedBy": [9] },
+    { "id": 12, "subject": "Task 12: Value-cache clone removal per the aliasing audit", "status": "completed", "blockedBy": [9] },
     { "id": 13, "subject": "Task 13: Phase B gate — windev full verification", "status": "pending", "blockedBy": [10, 11, 12] },
     { "id": 14, "subject": "Task 14: Wrap-up — deferred table closure, docs sweep, final review", "status": "pending", "blockedBy": [13] }
   ],

src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/DashboardPageBase.cs

@@ -1,4 +1,5 @@
 using Microsoft.AspNetCore.Components;
+using Microsoft.Extensions.Logging;
 
 namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Components;
 
@@ -31,6 +32,10 @@ public abstract class DashboardPageBase : ComponentBase, IAsyncDisposable
     [Inject]
     protected IDashboardSnapshotFeed SnapshotFeed { get; set; } = null!;
 
+    /// <summary>Logger used to report a snapshot subscription that ended or would not drain.</summary>
+    [Inject]
+    protected ILogger<DashboardPageBase>? Logger { get; set; }
+
     /// <summary>
     /// The most recent gateway metric snapshot. Synchronously seeded from
     /// <see cref="IDashboardSnapshotService.GetSnapshot"/> for the very first
@@ -62,9 +67,21 @@ public abstract class DashboardPageBase : ComponentBase, IAsyncDisposable
                 await _watchTask.WaitAsync(WatchDrainTimeout).ConfigureAwait(false);
             }
         }
+        catch (TimeoutException)
+        {
+            // Accepted limitation: the abandoned loop still holds its feed subscription, so
+            // the feed's idle gate stays open until it does unwind. There is no way to force
+            // a detach — the loop is parked on a dispatcher that is not draining — so the
+            // warning is the operator's only signal that a circuit teardown wedged.
+            Logger?.LogWarning(
+                "Dashboard page {Page} did not release its snapshot subscription within {Timeout}; "
+                + "the shared snapshot feed stays active until it unwinds.",
+                GetType().Name,
+                WatchDrainTimeout);
+        }
         catch
         {
-            // Disposal-time errors (including a drain timeout) are best-effort.
+            // Other disposal-time errors are best-effort.
         }
 
         _watchCancellation.Dispose();
@@ -87,10 +104,15 @@ public abstract class DashboardPageBase : ComponentBase, IAsyncDisposable
         {
             // The page is going away.
         }
-        catch
+        catch (Exception error)
         {
-            // The feed is best-effort: the last rendered snapshot stays on screen and
-            // the snapshot service keeps serving GetSnapshot() for the next page load.
+            // The feed is best-effort: the last rendered snapshot stays on screen and the
+            // snapshot service keeps serving GetSnapshot() for the next page load. Logged
+            // once here, on the way out of the loop — never per snapshot.
+            Logger?.LogWarning(
+                error,
+                "Live snapshot updates ended for dashboard page {Page}; it keeps the last rendered snapshot.",
+                GetType().Name);
         }
     }
 }

src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/AlarmsPage.razor

@@ -207,8 +207,11 @@
                         continue;
                     }
 
-                    _providerStatus = DashboardAlarmProviderStatus.FromFeed(message);
-                    await InvokeAsync(StateHasChanged).ConfigureAwait(false);
+                    await InvokeAsync(() =>
+                    {
+                        _providerStatus = DashboardAlarmProviderStatus.FromFeed(message);
+                        StateHasChanged();
+                    }).ConfigureAwait(false);
                 }
             }
             catch (OperationCanceledException)

src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/SessionDetailsPage.razor

@@ -5,7 +5,7 @@
 @using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs
 @inject AuthenticationStateProvider AuthenticationStateProvider
 @inject IDashboardSessionAdminService SessionAdminService
-@inject IDashboardEventBroadcaster EventBroadcaster
+@inject IDashboardSessionEventSubscriber EventSubscriber
 
 <PageTitle>Dashboard Session</PageTitle>
 
@@ -157,8 +157,18 @@ else
     private DashboardSessionSummary? CurrentSession => Snapshot?.Sessions.FirstOrDefault(session =>
         string.Equals(session.SessionId, SessionId, StringComparison.Ordinal));
 
+    // Upper bound on waiting for the event pump while detaching, mirroring
+    // DashboardPageBase's snapshot-watch drain: the pump marshals renders through the
+    // renderer's dispatcher and a detach can run on that same dispatcher, so the wait
+    // is bounded rather than unconditional.
+    private static readonly TimeSpan EventPumpDrainTimeout = TimeSpan.FromSeconds(5);
+
+    // Written only on the renderer's dispatcher (the lifecycle methods below), and read
+    // on it from inside the pump's dispatched callback — that pairing is what makes the
+    // stale-batch guard in PumpEventsAsync reliable.
     private IDashboardEventSubscription? _eventSubscription;
     private CancellationTokenSource? _eventPumpCancellation;
+    private Task? _eventPumpTask;
     private bool _eventsConnected;
     private string? _subscribedSessionId;
     private readonly LinkedList<MxEvent> _recentEvents = new();
@@ -180,17 +190,16 @@ else
         CanManage = SessionAdminService.CanManage(authenticationState.User);
     }
 
-    protected override Task OnParametersSetAsync()
+    protected override async Task OnParametersSetAsync()
     {
-        // Attach/detach are synchronous now that the feed is in-process; the override
-        // stays on the async lifecycle member so the base class's contract is untouched.
         if (!string.Equals(_subscribedSessionId, SessionId, StringComparison.Ordinal))
         {
-            DetachEvents();
+            // Deliberately no ConfigureAwait(false): the resumption must stay on the
+            // renderer's dispatcher so the new subscription is published to
+            // _eventSubscription from the same thread the pump's guard reads it on.
+            await DetachEventsAsync();
             AttachEvents();
         }
-
-        return Task.CompletedTask;
     }
 
     private PendingConfirm? PendingAction { get; set; }
@@ -266,27 +275,31 @@ else
         string ConfirmButtonClass,
         Func<System.Security.Claims.ClaimsPrincipal, Task<DashboardSessionAdminResult>> Action);
 
-    // The dashboard runs in the same process as the broadcaster, so this page reads
+    // The dashboard runs in the same process as the event mirror, so this page reads
     // the session's mirrored events straight from it. It used to open a loopback
     // SignalR connection to /hubs/events — mint a hub token, negotiate, hold a
     // WebSocket, serialize every event — to reach data already sitting in memory.
-    // The subscription still registers with EventsHubViewerRegistry, so the
-    // broadcaster's "nobody is watching" gate keeps working for both audiences.
+    // IDashboardSessionEventSubscriber resolves to the same singleton that serves
+    // IDashboardEventBroadcaster, and the subscription registers with
+    // EventsHubViewerRegistry, so the "nobody is watching" gate keeps working for
+    // both audiences.
     // ACL posture is unchanged from the hub path: any dashboard Viewer may watch
     // any session (SEC-25 tracks the per-session ACL for both seams).
     private void AttachEvents()
     {
-        if (string.IsNullOrWhiteSpace(SessionId) || EventBroadcaster is not IDashboardSessionEventSubscriber subscriber)
+        if (string.IsNullOrWhiteSpace(SessionId))
         {
             return;
         }
 
-        _eventSubscription = subscriber.Subscribe(SessionId);
+        _eventSubscription = EventSubscriber.Subscribe(SessionId);
         _eventPumpCancellation = new CancellationTokenSource();
         _eventsConnected = true;
         _subscribedSessionId = SessionId;
 
-        _ = PumpEventsAsync(_eventSubscription, _eventPumpCancellation.Token);
+        // Deliberately not awaited: the pump runs for as long as the page watches this
+        // session and is cancelled and drained by DetachEventsAsync.
+        _eventPumpTask = PumpEventsAsync(_eventSubscription, _eventPumpCancellation.Token);
     }
 
     private async Task PumpEventsAsync(IDashboardEventSubscription subscription, CancellationToken cancellationToken)
@@ -311,6 +324,17 @@ else
 
                 await InvokeAsync(() =>
                 {
+                    // The batch was read before this callback was dispatched, and a
+                    // session switch can land in between. Rendering it then would show
+                    // the previous session's events under the new session's heading, so
+                    // a batch whose subscription is no longer the live one is dropped.
+                    // Safe as an unsynchronized read: _eventSubscription is written on
+                    // this same dispatcher.
+                    if (!ReferenceEquals(_eventSubscription, subscription))
+                    {
+                        return;
+                    }
+
                     foreach (MxEvent mxEvent in batch)
                     {
                         _recentEvents.AddFirst(mxEvent);
@@ -331,26 +355,43 @@ else
         }
         catch (ObjectDisposedException)
         {
-            // The renderer went away while a batch was being dispatched.
+            // Either the renderer went away mid-dispatch, or the drain below timed out
+            // and disposed the cancellation source this loop is still reading.
         }
     }
 
-    private void DetachEvents()
+    private async Task DetachEventsAsync()
     {
         IDashboardEventSubscription? subscription = _eventSubscription;
         CancellationTokenSource? cancellation = _eventPumpCancellation;
+        Task? pump = _eventPumpTask;
         _eventSubscription = null;
         _eventPumpCancellation = null;
+        _eventPumpTask = null;
         _eventsConnected = false;
         _subscribedSessionId = null;
         _recentEvents.Clear();
 
-        // Cancel first so the pump stops touching the renderer, then dispose the
-        // subscription — that is what releases the viewer registration and lets the
-        // broadcaster go back to skipping mirror work for this session.
+        // Cancel and drop the subscription before draining. Disposing it releases the
+        // viewer registration — the whole point of the gate — and completes the channel,
+        // so the pump has an exit even if cancellation is missed.
         cancellation?.Cancel();
-        cancellation?.Dispose();
         subscription?.Dispose();
+
+        try
+        {
+            if (pump is not null)
+            {
+                await pump.WaitAsync(EventPumpDrainTimeout);
+            }
+        }
+        catch
+        {
+            // Detach-time errors (including a drain timeout) are best-effort.
+        }
+
+        // Disposed after the drain so the pump is no longer reading the token.
+        cancellation?.Dispose();
     }
 
     private static string EventStatusLabel(MxEvent evt)
@@ -362,7 +403,7 @@ else
 
     public new async ValueTask DisposeAsync()
     {
-        DetachEvents();
+        await DetachEventsAsync();
         await base.DisposeAsync().ConfigureAwait(false);
     }
 }

src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs

@@ -51,7 +51,17 @@ public static class DashboardServiceCollectionExtensions
         // Singleton: EventsHub instances are transient (one per hub invocation), so the
         // subscriber bookkeeping they share with the broadcaster must outlive them.
         services.AddSingleton<Hubs.EventsHubViewerRegistry>();
-        services.AddSingleton<Hubs.IDashboardEventBroadcaster, Hubs.DashboardEventBroadcaster>();
+
+        // One instance behind two interfaces, registered concretely and forwarded: the
+        // publish side (IDashboardEventBroadcaster, driven by the session pipeline) and
+        // the in-process subscribe side (IDashboardSessionEventSubscriber, used by the
+        // session-details page) share subscriber bookkeeping, so resolving them to two
+        // instances would leave the page subscribed to a mirror nobody publishes to.
+        services.AddSingleton<Hubs.DashboardEventBroadcaster>();
+        services.AddSingleton<Hubs.IDashboardEventBroadcaster>(
+            static provider => provider.GetRequiredService<Hubs.DashboardEventBroadcaster>());
+        services.AddSingleton<Hubs.IDashboardSessionEventSubscriber>(
+            static provider => provider.GetRequiredService<Hubs.DashboardEventBroadcaster>());
         services.AddSingleton<Hubs.DashboardSnapshotHubConnectionCounter>();
         services.AddHostedService<Hubs.DashboardSnapshotPublisher>();
         services.AddHostedService<Hubs.AlarmsHubPublisher>();

src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs

@@ -12,18 +12,31 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
 /// snapshot cost by the number of open pages.
 /// </summary>
 /// <remarks>
-/// The pump is idle-gated: it starts when the subscriber count goes 0 → 1 and is cancelled
-/// and awaited when it goes 1 → 0, so an unwatched gateway runs no timer and builds no
-/// snapshots. Successive pumps are chained through <c>_pumpTask</c>, so a rapid
+/// <para>
+/// The pump is idle-gated: it starts when the first subscriber arrives and is cancelled and
+/// awaited when the last subscriber <em>of the live generation</em> leaves, so an unwatched
+/// gateway runs no timer and builds no snapshots. Successive pumps are chained through
+/// <c>_pumpTask</c>, so a rapid
 /// unsubscribe/resubscribe restarts a fresh pump without ever running two enumerations at
 /// once.
+/// </para>
+/// <para>
+/// Every subscriber is tagged with the pump generation it joined under, and a dying pump only
+/// ever detaches its own generation. A pump ends its generation the instant its source fails
+/// or completes — before the (possibly slow) enumerator disposal — so a subscriber arriving
+/// while a pump unwinds starts a fresh generation instead of silently attaching to a dead
+/// pump that is about to detach everybody and leave nobody watching.
+/// </para>
 /// </remarks>
 public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
 {
+    /// <summary>Generation value meaning "no pump is accepting subscribers".</summary>
+    private const long NoGeneration = 0;
+
     private readonly IDashboardSnapshotService _snapshotService;
     private readonly ILogger<DashboardSnapshotFeed> _logger;
     private readonly object _gate = new();
-    private readonly List<Channel<DashboardSnapshot>> _subscribers = [];
+    private readonly List<Subscription> _subscribers = [];
 
     /// <summary>
     /// The most recent pump, completed while idle. A starting pump awaits its predecessor
@@ -31,9 +44,15 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
     /// </summary>
     private Task _pumpTask = Task.CompletedTask;
 
-    /// <summary>Cancellation for the live pump; null when no pump is running or one is being torn down.</summary>
+    /// <summary>Cancellation for the live pump; null when no generation is accepting subscribers.</summary>
     private CancellationTokenSource? _pumpCancellation;
 
+    /// <summary>The generation new subscribers join, or <see cref="NoGeneration"/> when no pump is live.</summary>
+    private long _generation = NoGeneration;
+
+    /// <summary>Last generation handed out; only ever incremented under <c>_gate</c>.</summary>
+    private long _lastGeneration = NoGeneration;
+
     /// <summary>Initializes a new instance of the <see cref="DashboardSnapshotFeed"/> class.</summary>
     /// <param name="snapshotService">Snapshot source to multicast.</param>
     /// <param name="logger">Optional logger for pump faults.</param>
@@ -60,7 +79,7 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
                 SingleWriter = false,
             });
 
-        Subscribe(channel);
+        Subscription subscription = Subscribe(channel);
         try
         {
             await foreach (DashboardSnapshot snapshot in channel.Reader
@@ -74,47 +93,55 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
         {
             // Untokened on purpose: teardown must run to completion even when this
             // subscriber is unwinding because its own token fired.
-            await UnsubscribeAsync(channel).ConfigureAwait(false);
+            await UnsubscribeAsync(subscription).ConfigureAwait(false);
         }
     }
 
-    private void Subscribe(Channel<DashboardSnapshot> channel)
+    private Subscription Subscribe(Channel<DashboardSnapshot> channel)
     {
         lock (_gate)
         {
-            _subscribers.Add(channel);
-            if (_subscribers.Count != 1)
-            {
-                return;
-            }
-
-            CancellationTokenSource cancellation = new();
-            Task previous = _pumpTask;
-            _pumpCancellation = cancellation;
-
-            // Task.Run, not a direct call: an async iterator runs synchronously up to its
-            // first suspension, and the first pull of the underlying watch can read the API
-            // key table. That must not run on the subscribing circuit's thread, let alone
-            // while this lock is held.
-            _pumpTask = Task.Run(() => PumpAsync(previous, cancellation, cancellation.Token));
+            // A live generation is joined; otherwise this subscriber starts one. Keying on
+            // "is a generation live" rather than "is this the first subscriber" is what makes
+            // a subscriber arriving while a pump unwinds start a fresh pump for itself.
+            long generation = _pumpCancellation is null ? StartPumpLocked() : _generation;
+            Subscription subscription = new(channel, generation);
+            _subscribers.Add(subscription);
+            return subscription;
         }
     }
 
-    private async Task UnsubscribeAsync(Channel<DashboardSnapshot> channel)
+    private async Task UnsubscribeAsync(Subscription subscription)
     {
         CancellationTokenSource? cancellation;
         Task pump;
         lock (_gate)
         {
-            if (!_subscribers.Remove(channel) || _subscribers.Count != 0)
+            if (!_subscribers.Remove(subscription))
             {
-                // Either the pump already dropped this channel (it completed or faulted
-                // and reset itself), or other viewers are still watching.
+                // The pump already detached this subscription (it completed or faulted).
+                return;
+            }
+
+            if (subscription.Generation != _generation)
+            {
+                // This viewer belonged to a generation that has already ended. The live
+                // pump — if there is one — serves other viewers and must not be cancelled
+                // on their behalf; the dying pump is stopping under its own steam.
+                return;
+            }
+
+            if (HasSubscribersLocked(_generation))
+            {
+                // Other viewers are still watching the live generation. Counting the whole
+                // list here would be wrong: subscribers of an ending generation linger in it
+                // until that pump's Reset runs, and they must not hold the idle gate open.
                 return;
             }
 
             cancellation = _pumpCancellation;
             _pumpCancellation = null;
+            _generation = NoGeneration;
             pump = _pumpTask;
         }
 
@@ -124,7 +151,7 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
         }
         catch (ObjectDisposedException)
         {
-            // The pump reset itself and disposed its own cancellation source first.
+            // The pump ended on its own and disposed its cancellation source first.
         }
 
         try
@@ -138,33 +165,106 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
         }
     }
 
-    private async Task PumpAsync(Task previous, CancellationTokenSource cancellation, CancellationToken cancellationToken)
+    /// <summary>Reports whether any subscriber is still being served by a generation.</summary>
+    /// <param name="generation">The generation to look for.</param>
+    /// <returns>True when at least one subscriber carries that generation.</returns>
+    private bool HasSubscribersLocked(long generation)
+    {
+        foreach (Subscription subscriber in _subscribers)
+        {
+            if (subscriber.Generation == generation)
+            {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    /// <summary>
+    /// Starts a pump generation. Must be called while holding <c>_gate</c>; the caller adds
+    /// the subscribers that belong to the returned generation.
+    /// </summary>
+    /// <returns>The new generation identifier.</returns>
+    private long StartPumpLocked()
+    {
+        long generation = ++_lastGeneration;
+        CancellationTokenSource cancellation = new();
+        Task previous = _pumpTask;
+        _generation = generation;
+        _pumpCancellation = cancellation;
+
+        // Task.Run, not a direct call: an async iterator runs synchronously up to its
+        // first suspension, and the first pull of the underlying watch can read the API
+        // key table. That must not run on the subscribing circuit's thread, let alone
+        // while this lock is held.
+        _pumpTask = Task.Run(() => PumpAsync(generation, previous, cancellation, cancellation.Token));
+        return generation;
+    }
+
+    private async Task PumpAsync(
+        long generation,
+        Task previous,
+        CancellationTokenSource cancellation,
+        CancellationToken cancellationToken)
     {
         try
         {
             // Never overlap with the enumeration this pump replaces.
             await previous.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
 
-            await foreach (DashboardSnapshot snapshot in _snapshotService
+            // Enumerated by hand rather than with await foreach so the generation can be
+            // ended the moment the source fails or completes — await foreach would run the
+            // enumerator's disposal first, and a subscriber arriving during that disposal
+            // would join a generation that is already doomed.
+            IAsyncEnumerator<DashboardSnapshot> snapshots = _snapshotService
                 .WatchSnapshotsAsync(cancellationToken)
-                .ConfigureAwait(false))
+                .GetAsyncEnumerator(cancellationToken);
+            try
             {
-                Broadcast(snapshot);
+                while (true)
+                {
+                    bool moved;
+                    try
+                    {
+                        moved = await snapshots.MoveNextAsync().ConfigureAwait(false);
+                    }
+                    catch
+                    {
+                        EndGeneration(generation);
+                        throw;
+                    }
+
+                    if (!moved)
+                    {
+                        EndGeneration(generation);
+                        break;
+                    }
+
+                    Broadcast(generation, snapshots.Current);
+                }
+            }
+            finally
+            {
+                await snapshots.DisposeAsync().ConfigureAwait(false);
             }
 
-            // The source completed on its own; hand the completion to the subscribers
-            // and re-arm so the next one starts a fresh enumeration.
-            Reset(cancellation, error: null);
+            // The source completed on its own; hand the completion to this generation's
+            // subscribers and re-arm so the next one starts a fresh enumeration.
+            Reset(generation, error: null);
         }
         catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
         {
-            // Last subscriber left: the unsubscribing caller already detached the
-            // channels and cleared the pump state.
+            // The production DashboardSnapshotService swallows cancellation and yield-breaks,
+            // so normal teardown exits through the fall-through above (with an ownership-checked
+            // Reset that finds no subscribers); an implementation that propagates the token
+            // instead exits here. Both shapes end the generation exactly once.
+            EndGeneration(generation);
         }
         catch (Exception error)
         {
             _logger.LogWarning(error, "Dashboard snapshot feed stopped; the next subscriber restarts it.");
-            Reset(cancellation, error);
+            Reset(generation, error);
         }
         finally
         {
@@ -172,45 +272,101 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
         }
     }
 
-    private void Broadcast(DashboardSnapshot snapshot)
+    private void Broadcast(long generation, DashboardSnapshot snapshot)
     {
         lock (_gate)
         {
-            foreach (Channel<DashboardSnapshot> subscriber in _subscribers)
+            foreach (Subscription subscriber in _subscribers)
             {
+                if (subscriber.Generation != generation)
+                {
+                    continue;
+                }
+
                 // Bounded/DropOldest: always accepted unless the channel is completed.
-                subscriber.Writer.TryWrite(snapshot);
+                subscriber.Channel.Writer.TryWrite(snapshot);
             }
         }
     }
 
     /// <summary>
-    /// Detaches every subscriber and clears the pump state so the next subscriber starts a
-    /// new enumeration. The detached subscribers observe <paramref name="error"/> (or a
-    /// clean end of stream) from their own <c>WatchAsync</c>.
+    /// Stops handing <paramref name="generation"/> to new subscribers. Called the instant a
+    /// pump's source fails or completes, before its enumerator is disposed.
     /// </summary>
-    /// <param name="cancellation">The calling pump's cancellation source, used as its ownership token.</param>
-    /// <param name="error">Failure to surface, or null when the source completed cleanly.</param>
-    private void Reset(CancellationTokenSource cancellation, Exception? error)
+    /// <param name="generation">The generation that has ended.</param>
+    private void EndGeneration(long generation)
     {
-        Channel<DashboardSnapshot>[] detached;
         lock (_gate)
         {
-            if (!ReferenceEquals(_pumpCancellation, cancellation))
-            {
-                // A newer pump (or an in-flight teardown) owns the state now: its
-                // subscribers must not be detached by this pump's exit.
-                return;
-            }
-
-            detached = _subscribers.ToArray();
-            _subscribers.Clear();
-            _pumpCancellation = null;
-        }
-
-        foreach (Channel<DashboardSnapshot> subscriber in detached)
-        {
-            subscriber.Writer.TryComplete(error);
+            EndGenerationLocked(generation);
         }
     }
+
+    /// <summary>Clears the live-pump state if <paramref name="generation"/> still owns it.</summary>
+    /// <param name="generation">The generation that has ended.</param>
+    private void EndGenerationLocked(long generation)
+    {
+        if (_generation != generation)
+        {
+            return;
+        }
+
+        _generation = NoGeneration;
+        _pumpCancellation = null;
+    }
+
+    /// <summary>
+    /// Detaches the subscribers of a finished generation and re-arms the feed. Subscribers of
+    /// any other generation are left alone — they belong to a pump that is still running (or
+    /// about to), so a dying pump must not take them down with it.
+    /// </summary>
+    /// <param name="generation">The generation whose subscribers are being detached.</param>
+    /// <param name="error">Failure to surface, or null when the source completed cleanly.</param>
+    private void Reset(long generation, Exception? error)
+    {
+        List<Channel<DashboardSnapshot>> detached = [];
+        lock (_gate)
+        {
+            for (int index = _subscribers.Count - 1; index >= 0; index--)
+            {
+                if (_subscribers[index].Generation != generation)
+                {
+                    continue;
+                }
+
+                detached.Add(_subscribers[index].Channel);
+                _subscribers.RemoveAt(index);
+            }
+
+            EndGenerationLocked(generation);
+
+            if (_subscribers.Count > 0 && _pumpCancellation is null)
+            {
+                // Belt and braces: subscribers left with no live pump would be frozen for
+                // good, because only a subscriber that finds no generation starts one.
+                long restarted = StartPumpLocked();
+                foreach (Subscription subscriber in _subscribers)
+                {
+                    subscriber.Generation = restarted;
+                }
+            }
+        }
+
+        foreach (Channel<DashboardSnapshot> channel in detached)
+        {
+            channel.Writer.TryComplete(error);
+        }
+    }
+
+    /// <summary>One viewer's delivery channel plus the pump generation serving it.</summary>
+    /// <param name="channel">Delivery channel for this viewer.</param>
+    /// <param name="generation">Pump generation this viewer joined under.</param>
+    private sealed class Subscription(Channel<DashboardSnapshot> channel, long generation)
+    {
+        /// <summary>Gets the viewer's delivery channel.</summary>
+        public Channel<DashboardSnapshot> Channel { get; } = channel;
+
+        /// <summary>Gets or sets the pump generation currently serving this viewer.</summary>
+        public long Generation { get; set; } = generation;
+    }
 }

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardHubsRegistrationTests.cs

@@ -45,4 +45,23 @@ public sealed class DashboardHubsRegistrationTests
             .GetRequiredService<DashboardHubConnectionFactory>();
         Assert.NotNull(factory);
     }
+
+    /// <summary>
+    /// The publish and in-process subscribe faces of the event mirror must resolve to
+    /// one instance: two would leave the session-details page reading a mirror the
+    /// session pipeline never publishes to, and the viewer gate would never open.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Build_WhenDashboardEnabled_ResolvesBothEventMirrorInterfacesToOneInstance()
+    {
+        await using WebApplication app = GatewayApplication.Build([]);
+
+        IDashboardEventBroadcaster broadcaster = app.Services.GetRequiredService<IDashboardEventBroadcaster>();
+        IDashboardSessionEventSubscriber subscriber = app.Services
+            .GetRequiredService<IDashboardSessionEventSubscriber>();
+
+        Assert.Same(broadcaster, subscriber);
+        Assert.Same(app.Services.GetRequiredService<DashboardEventBroadcaster>(), broadcaster);
+    }
 }

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs

@@ -201,6 +201,125 @@ public sealed class DashboardSnapshotFeedTests
         await DrainAsync(second, secondMove);
     }
 
+    /// <summary>
+    /// The race the generation tagging exists for: a page subscribes in the window between
+    /// the source failing and the dying pump detaching its subscribers. Without generations
+    /// the newcomer joined the doomed pump, was detached with its error, and no pump ever
+    /// restarted (only a first subscriber started one) — that page froze for good.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WatchAsync_WhenASubscriberJoinsWhileAFaultedPumpUnwinds_IsServedByAFreshPump()
+    {
+        FakeSnapshotService service = new();
+        service.HoldDisposal();
+        DashboardSnapshotFeed feed = new(service);
+
+        using CancellationTokenSource firstCancellation = new();
+        IAsyncEnumerator<DashboardSnapshot> first =
+            feed.WatchAsync(firstCancellation.Token).GetAsyncEnumerator(firstCancellation.Token);
+        Task<bool> firstMove = first.MoveNextAsync().AsTask();
+        await WaitUntilAsync(() => service.EnumerationCount >= 1);
+
+        service.Fault(new InvalidOperationException("simulated snapshot source failure"));
+
+        // The pump has observed the failure and is parked disposing the enumerator — the
+        // exact window in which a page used to attach itself to a doomed pump.
+        await service.DisposalReached.WaitAsync(TestTimeout);
+
+        using CancellationTokenSource secondCancellation = new();
+        IAsyncEnumerator<DashboardSnapshot> second =
+            feed.WatchAsync(secondCancellation.Token).GetAsyncEnumerator(secondCancellation.Token);
+
+        // An async iterator body runs synchronously up to its first await, so the
+        // subscription is registered by the time MoveNextAsync hands back its task.
+        Task<bool> secondMove = second.MoveNextAsync().AsTask();
+
+        service.ReleaseDisposal();
+
+        // The subscriber that was there when the source broke still learns about it...
+        InvalidOperationException failure =
+            await Assert.ThrowsAsync<InvalidOperationException>(() => firstMove.WaitAsync(TestTimeout));
+        Assert.Equal("simulated snapshot source failure", failure.Message);
+        await first.DisposeAsync();
+
+        // ...and the one that joined mid-unwind is served by a restarted enumeration
+        // instead of inheriting the failure.
+        await WaitUntilAsync(() => service.EnumerationCount >= 2);
+        await PushUntilAsync(service, secondMove);
+        Assert.True(await secondMove.WaitAsync(TestTimeout));
+        Assert.StartsWith("push-", second.Current.GatewayVersion, StringComparison.Ordinal);
+
+        await secondCancellation.CancelAsync();
+        await DrainAsync(second, secondMove);
+    }
+
+    /// <summary>
+    /// The idle gate is per generation, not per subscriber count. A viewer that joins while a
+    /// pump unwinds starts a new generation, and the old generation's viewers linger in the
+    /// list until that pump's reset runs — so a global "is the list empty" check let the new
+    /// viewer leave without cancelling the generation it had just started, leaving a pump
+    /// enumerating the snapshot source with nobody watching it.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WatchAsync_WhenAJoinerLeavesWhileAnOldGenerationLingers_LeavesNoPumpRunning()
+    {
+        FakeSnapshotService service = new();
+        service.HoldDisposal();
+        DashboardSnapshotFeed feed = new(service);
+
+        using CancellationTokenSource oldCancellation = new();
+        IAsyncEnumerator<DashboardSnapshot> oldSubscriber =
+            feed.WatchAsync(oldCancellation.Token).GetAsyncEnumerator(oldCancellation.Token);
+        Task<bool> oldMove = oldSubscriber.MoveNextAsync().AsTask();
+        await WaitUntilAsync(() => service.EnumerationCount >= 1);
+
+        service.Fault(new InvalidOperationException("simulated snapshot source failure"));
+        await service.DisposalReached.WaitAsync(TestTimeout);
+
+        // Joins mid-unwind (starting a fresh generation) and leaves again before the dying
+        // pump has detached the subscriber that is still lingering in the list.
+        using CancellationTokenSource joinerCancellation = new();
+        IAsyncEnumerator<DashboardSnapshot> joiner =
+            feed.WatchAsync(joinerCancellation.Token).GetAsyncEnumerator(joinerCancellation.Token);
+        Task<bool> joinerMove = joiner.MoveNextAsync().AsTask();
+        await joinerCancellation.CancelAsync();
+
+        // The joiner's unwind is a continuation of its cancelled channel read; give it time to
+        // run its unsubscribe before the dying pump is released. Only the interleaving depends
+        // on this delay — the assertions below hold either way.
+        await Task.Delay(TimeSpan.FromMilliseconds(150));
+
+        service.ReleaseDisposal();
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() => oldMove.WaitAsync(TestTimeout));
+        await oldSubscriber.DisposeAsync();
+
+        // Completes only once the joiner's unsubscribe has awaited its generation's pump.
+        await DrainAsync(joiner, joinerMove);
+
+        // Nobody is watching, so nothing may consume a snapshot: a surviving pump would drain
+        // this push within its first read.
+        service.Push(CreateSnapshot("orphan-check"));
+        await Task.Delay(TimeSpan.FromMilliseconds(150));
+        Assert.Equal(1, service.PendingPushCount);
+
+        // ...and the next viewer still starts cleanly, picking up the queued snapshot.
+        int enumerationsBefore = service.EnumerationCount;
+        using CancellationTokenSource nextCancellation = new();
+        IAsyncEnumerator<DashboardSnapshot> next =
+            feed.WatchAsync(nextCancellation.Token).GetAsyncEnumerator(nextCancellation.Token);
+        Task<bool> nextMove = next.MoveNextAsync().AsTask();
+
+        await WaitUntilAsync(() => service.EnumerationCount > enumerationsBefore);
+        Assert.True(await nextMove.WaitAsync(TestTimeout));
+        Assert.Equal("orphan-check", next.Current.GatewayVersion);
+
+        await nextCancellation.CancelAsync();
+        await DrainAsync(next, nextMove);
+    }
+
     /// <summary>Builds a snapshot whose version string identifies it in assertions.</summary>
     /// <param name="version">Identity marker carried in <c>GatewayVersion</c>.</param>
     /// <returns>A snapshot carrying the supplied identity marker.</returns>
@@ -282,13 +401,25 @@ public sealed class DashboardSnapshotFeedTests
     private sealed class FakeSnapshotService : IDashboardSnapshotService
     {
         private readonly Channel<object> _pushes = Channel.CreateUnbounded<object>();
+        private readonly TaskCompletionSource _disposalReached = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly TaskCompletionSource _disposalRelease = new(TaskCreationOptions.RunContinuationsAsynchronously);
         private int _enumerationCount;
         private int _completedEnumerationCount;
         private volatile bool _lastEnumerationWasCancelled;
+        private volatile bool _holdDisposal;
 
         /// <summary>Gets the number of times the feed started enumerating this source.</summary>
         public int EnumerationCount => Volatile.Read(ref _enumerationCount);
 
+        /// <summary>Gets a task that completes when a held enumerator disposal is reached.</summary>
+        public Task DisposalReached => _disposalReached.Task;
+
+        /// <summary>
+        /// Gets the number of queued snapshots no enumeration has taken yet. A live pump
+        /// drains this even with nobody watching, so a stable count proves the feed is idle.
+        /// </summary>
+        public int PendingPushCount => _pushes.Reader.Count;
+
         /// <summary>Gets the number of enumerations that have finished (cancelled, faulted, or completed).</summary>
         public int CompletedEnumerationCount => Volatile.Read(ref _completedEnumerationCount);
 
@@ -303,11 +434,34 @@ public sealed class DashboardSnapshotFeedTests
         /// <param name="error">Exception to throw from the enumeration.</param>
         public void Fault(Exception error) => _pushes.Writer.TryWrite(error);
 
+        /// <summary>
+        /// Parks enumerator disposal until <see cref="ReleaseDisposal"/>, which holds the pump
+        /// in the window between observing the source's failure and detaching its subscribers.
+        /// </summary>
+        public void HoldDisposal() => _holdDisposal = true;
+
+        /// <summary>Releases a held enumerator disposal.</summary>
+        public void ReleaseDisposal() => _disposalRelease.TrySetResult();
+
         /// <inheritdoc />
         public DashboardSnapshot GetSnapshot() => CreateSnapshot("current");
 
         /// <inheritdoc />
-        public async IAsyncEnumerable<DashboardSnapshot> WatchSnapshotsAsync(
+        public IAsyncEnumerable<DashboardSnapshot> WatchSnapshotsAsync(CancellationToken cancellationToken)
+            => new GatedEnumerable(this);
+
+        private async Task OnDisposingAsync()
+        {
+            if (!_holdDisposal)
+            {
+                return;
+            }
+
+            _disposalReached.TrySetResult();
+            await _disposalRelease.Task.ConfigureAwait(false);
+        }
+
+        private async IAsyncEnumerable<DashboardSnapshot> EnumerateAsync(
             [EnumeratorCancellation] CancellationToken cancellationToken)
         {
             Interlocked.Increment(ref _enumerationCount);
@@ -329,5 +483,40 @@ public sealed class DashboardSnapshotFeedTests
                 Interlocked.Increment(ref _completedEnumerationCount);
             }
         }
+
+        /// <summary>
+        /// Wraps the iterator so disposal is a control point of its own: the feed ends a pump
+        /// generation when MoveNextAsync fails, which is strictly before this disposal runs.
+        /// </summary>
+        /// <param name="owner">The fake whose enumeration is being wrapped.</param>
+        private sealed class GatedEnumerable(FakeSnapshotService owner) : IAsyncEnumerable<DashboardSnapshot>
+        {
+            /// <summary>Creates a gated enumerator over the fake's enumeration.</summary>
+            /// <param name="cancellationToken">Token to observe for cancellation.</param>
+            /// <returns>The gated enumerator.</returns>
+            public IAsyncEnumerator<DashboardSnapshot> GetAsyncEnumerator(
+                CancellationToken cancellationToken = default)
+                => new GatedEnumerator(owner, owner.EnumerateAsync(cancellationToken).GetAsyncEnumerator(cancellationToken));
+        }
+
+        private sealed class GatedEnumerator(
+            FakeSnapshotService owner,
+            IAsyncEnumerator<DashboardSnapshot> inner) : IAsyncEnumerator<DashboardSnapshot>
+        {
+            /// <summary>Gets the current snapshot.</summary>
+            public DashboardSnapshot Current => inner.Current;
+
+            /// <summary>Advances the wrapped enumeration.</summary>
+            /// <returns>A task that yields whether another snapshot is available.</returns>
+            public ValueTask<bool> MoveNextAsync() => inner.MoveNextAsync();
+
+            /// <summary>Parks while the fake holds disposal, then disposes the wrapped enumeration.</summary>
+            /// <returns>A task that represents the asynchronous operation.</returns>
+            public async ValueTask DisposeAsync()
+            {
+                await owner.OnDisposingAsync().ConfigureAwait(false);
+                await inner.DisposeAsync().ConfigureAwait(false);
+            }
+        }
     }
 }

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

@@ -377,6 +377,14 @@ public sealed class WorkerFrameProtocolTests
     /// Verifies the writer coalesces the flush across a batch of frames drained together: four frames
     /// queued behind an in-progress write drain in a single pass and share one FlushAsync, not four.
     /// Every frame still reaches the wire intact.
+    /// <para>
+    /// The burst is all-event on purpose. The control-frame completion decoupling made the drain flush
+    /// at each control-to-event boundary, so a pass that mixes classes legitimately pays one flush per
+    /// class run; the property
+    /// worth pinning is that a run of same-class frames — the event hot path — still costs exactly one
+    /// flush no matter how many frames drain together. The mixed shape has its own count assertion in
+    /// <see cref="DrainPass_MixedClasses_FlushesControlRunBeforeWritingEvents"/>.
+    /// </para>
     /// </summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
     [Fact]
@@ -387,7 +395,7 @@ public sealed class WorkerFrameProtocolTests
         WorkerFrameWriter writer = new(stream, options);
 
         // A blocked first write occupies the writer and holds the lock while more frames queue behind it.
-        Task firstWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
+        Task firstWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
         await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
 
         Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
@@ -410,6 +418,179 @@ public sealed class WorkerFrameProtocolTests
         }
     }
 
+    /// <summary>
+    /// Control-frame completion decoupling. A control frame's delivery point must not be charged for
+    /// the event backlog behind it. The priority scheduler already wrote control <em>bytes</em> first,
+    /// but a frame counts as
+    /// delivered only once flushed, and the pass deferred its single flush — and every completion —
+    /// until after the events. The drain now flushes at the control-to-event boundary: with two control
+    /// frames written and the first event write blocked inside the stream, the flush that closes out the
+    /// control run has already run, so the heartbeat or reply is on the pipe rather than waiting behind
+    /// the batch. Exactly two flushes for the pass — one per class run, not one per control frame.
+    /// <para>
+    /// The assertion is on the flush, not on the queued callers' returned tasks, because those tasks are
+    /// still gated by the write lock they lost to the drainer (see the latency contract on
+    /// <c>WorkerFrameWriter.WriteAsync</c>): the completion resolves at the boundary flush, but a
+    /// lock-race loser observes it only once the drainer releases the lock.
+    /// </para>
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DrainPass_MixedClasses_FlushesControlRunBeforeWritingEvents()
+    {
+        WorkerFrameProtocolOptions options = CreateOptions();
+        // Frame 1 (control) gates the pass open; frame 3 is the pass's first event write, which blocks
+        // so the boundary flush can be observed with the event batch still unwritten.
+        using GatedWriteStream stream = new(secondGateWriteIndex: 3);
+        WorkerFrameWriter writer = new(stream, options);
+
+        Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
+        await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
+
+        Task secondControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control);
+        Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
+        Task eventWrite2 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
+        await Task.Delay(50);
+
+        stream.ReleaseFirstWrite();
+
+        // The drain writes both control frames and is now blocked on the first event write.
+        await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted);
+
+        // The control run was flushed before the event batch was written — not after it.
+        Assert.Equal(1, stream.FlushCount);
+        Assert.False(eventWrite1.IsCompleted);
+        Assert.False(eventWrite2.IsCompleted);
+
+        stream.ReleaseSecondGateWrite();
+        await AwaitWithTimeoutAsync(
+            Task.WhenAll(firstControl, secondControl, eventWrite1, eventWrite2));
+
+        // One flush per class run: the control run, then the event run at the end of the pass.
+        Assert.Equal(2, stream.FlushCount);
+
+        stream.Position = 0;
+        WorkerFrameReader reader = new(stream, options);
+        WorkerEnvelope frame1 = await reader.ReadAsync();
+        WorkerEnvelope frame2 = await reader.ReadAsync();
+        WorkerEnvelope frame3 = await reader.ReadAsync();
+        WorkerEnvelope frame4 = await reader.ReadAsync();
+        Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase);
+        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, frame2.BodyCase);
+        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame3.BodyCase);
+        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame4.BodyCase);
+        Assert.Equal(stream.Length, stream.Position);
+    }
+
+    /// <summary>
+    /// Control-frame completion decoupling, the inverse guard. An event frame's completion boundary is
+    /// still the end-of-pass flush: a pure-event pass takes no boundary flush, so with two events
+    /// already written and the third
+    /// blocked mid-write, nothing has been flushed and no event can have been reported delivered. Only
+    /// a class transition may move a flush earlier — a plain event backlog may not.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DrainPass_PureEventRun_DoesNotFlushBeforeThePassEnds()
+    {
+        WorkerFrameProtocolOptions options = CreateOptions();
+        using GatedWriteStream stream = new(secondGateWriteIndex: 3);
+        WorkerFrameWriter writer = new(stream, options);
+
+        Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
+        await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
+
+        Task eventWrite2 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
+        Task eventWrite3 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
+        await Task.Delay(50);
+
+        stream.ReleaseFirstWrite();
+        await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted);
+
+        // Two event frames written, none flushed: no event frame's delivery point has been reached.
+        Assert.Equal(0, stream.FlushCount);
+        Assert.False(eventWrite1.IsCompleted);
+        Assert.False(eventWrite2.IsCompleted);
+
+        stream.ReleaseSecondGateWrite();
+        await AwaitWithTimeoutAsync(Task.WhenAll(eventWrite1, eventWrite2, eventWrite3));
+        Assert.Equal(1, stream.FlushCount);
+    }
+
+    /// <summary>
+    /// Control-frame completion decoupling. The boundary flush is charged per class run, not per
+    /// control frame: a pass carrying nothing but control frames still pays exactly one flush. Flushing
+    /// after every control frame
+    /// would reinstate the syscall-per-heartbeat cost WRK-12 removed.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DrainPass_PureControlRun_FlushesOnce()
+    {
+        WorkerFrameProtocolOptions options = CreateOptions();
+        using GatedWriteStream stream = new();
+        WorkerFrameWriter writer = new(stream, options);
+
+        Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
+        await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
+
+        Task secondControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
+        Task thirdControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control);
+        await Task.Delay(50);
+
+        stream.ReleaseFirstWrite();
+        await AwaitWithTimeoutAsync(Task.WhenAll(firstControl, secondControl, thirdControl));
+
+        Assert.Equal(1, stream.FlushCount);
+    }
+
+    /// <summary>
+    /// Control-frame completion decoupling, the new failure window. The boundary flush is a new place
+    /// the pipe can break with frames written but not yet delivered, so it must fail exactly like the
+    /// end-of-pass flush: every written control
+    /// frame fails, and so do the event frame the drain had already claimed off its queue (nothing else
+    /// would ever complete it) and every frame still queued, so no caller waits forever on a stream that
+    /// will not recover. The event bytes never reach the wire — the drain stops at the fault.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DrainPass_WhenBoundaryFlushFails_FailsWrittenClaimedAndQueuedFrames()
+    {
+        const string faultMessage = "boundary flush failed";
+        WorkerFrameProtocolOptions options = CreateOptions();
+        using FlushFaultingGatedStream stream = new(faultMessage);
+        WorkerFrameWriter writer = new(stream, options);
+
+        Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
+        await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
+
+        Task secondControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control);
+        Task claimedEvent = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
+        Task queuedEvent = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
+        await Task.Delay(50);
+
+        // The drain writes both control frames, claims the first event, and faults on the boundary flush.
+        stream.ReleaseFirstWrite();
+
+        // AwaitWithTimeoutAsync turns a frame nobody ever completes into a TimeoutException — a failed
+        // assertion rather than a hung test run.
+        foreach (Task write in new[] { firstControl, secondControl, claimedEvent, queuedEvent })
+        {
+            IOException failure = await Assert.ThrowsAsync<IOException>(
+                async () => await AwaitWithTimeoutAsync(write));
+            Assert.Equal(faultMessage, failure.Message);
+        }
+
+        // Only the control frames reached the wire; the claimed event was never written.
+        stream.Position = 0;
+        WorkerFrameReader reader = new(stream, options);
+        WorkerEnvelope frame1 = await reader.ReadAsync();
+        WorkerEnvelope frame2 = await reader.ReadAsync();
+        Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase);
+        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, frame2.BodyCase);
+        Assert.Equal(stream.Length, stream.Position);
+    }
+
     /// <summary>
     /// Verifies a per-frame rejection does not burn a sequence number (WRK-23). The sequence is a
     /// diagnostic counter, so a gap breaks nothing functionally — but an operator correlating a pipe
@@ -877,24 +1058,108 @@ public sealed class WorkerFrameProtocolTests
     }
 
     // A MemoryStream whose first WriteAsync blocks until released, so a test can queue additional frames
-    // behind an in-progress write and observe the writer's priority ordering.
+    // behind an in-progress write and observe the writer's priority ordering. A second, optional gate on
+    // a chosen write index lets a test stop a drain pass mid-flight — at a class boundary, say — and
+    // sample what the writer has already flushed while the rest of the pass is still unwritten.
     private sealed class GatedWriteStream : MemoryStream
     {
         private readonly SemaphoreSlim _release = new SemaphoreSlim(0);
+        private readonly SemaphoreSlim _secondGateRelease = new SemaphoreSlim(0);
         private readonly TaskCompletionSource<bool> _firstWriteStarted =
             new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly TaskCompletionSource<bool> _secondGateWriteStarted =
+            new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly int _secondGateWriteIndex;
         private int _writeCount;
         private int _flushCount;
 
+        /// <summary>Initializes a new instance of the GatedWriteStream class.</summary>
+        /// <param name="secondGateWriteIndex">
+        /// One-based index of a later write to block as well, or 0 (the default) to gate only the first
+        /// write. Write indexes start at 1, so 0 never matches.
+        /// </param>
+        public GatedWriteStream(int secondGateWriteIndex = 0)
+        {
+            _secondGateWriteIndex = secondGateWriteIndex;
+        }
+
         /// <summary>Gets a task that completes once the first <see cref="WriteAsync"/> call has started blocking.</summary>
         public Task FirstWriteStarted => _firstWriteStarted.Task;
 
+        /// <summary>Gets a task that completes once the second gated <see cref="WriteAsync"/> call has started blocking.</summary>
+        public Task SecondGateWriteStarted => _secondGateWriteStarted.Task;
+
         /// <summary>Gets the number of <see cref="FlushAsync"/> calls observed so far.</summary>
         public int FlushCount => Volatile.Read(ref _flushCount);
 
         /// <summary>Releases the first blocked write so it can complete.</summary>
         public void ReleaseFirstWrite() => _release.Release();
 
+        /// <summary>Releases the second gated write so it can complete.</summary>
+        public void ReleaseSecondGateWrite() => _secondGateRelease.Release();
+
+        /// <inheritdoc />
+        public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            int writeIndex = Interlocked.Increment(ref _writeCount);
+            if (writeIndex == 1)
+            {
+                _firstWriteStarted.TrySetResult(true);
+                await _release.WaitAsync(cancellationToken);
+            }
+            else if (writeIndex == _secondGateWriteIndex)
+            {
+                _secondGateWriteStarted.TrySetResult(true);
+                await _secondGateRelease.WaitAsync(cancellationToken);
+            }
+
+            await base.WriteAsync(buffer, offset, count, cancellationToken);
+        }
+
+        /// <inheritdoc />
+        public override Task FlushAsync(CancellationToken cancellationToken)
+        {
+            Interlocked.Increment(ref _flushCount);
+            return base.FlushAsync(cancellationToken);
+        }
+
+        /// <inheritdoc />
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                _release.Dispose();
+                _secondGateRelease.Dispose();
+            }
+
+            base.Dispose(disposing);
+        }
+    }
+
+    // A MemoryStream whose first write blocks until released and whose every FlushAsync throws, so a test
+    // can fault the class-boundary flush with control frames already written and an event frame already
+    // claimed off its queue.
+    private sealed class FlushFaultingGatedStream : MemoryStream
+    {
+        private readonly SemaphoreSlim _release = new SemaphoreSlim(0);
+        private readonly TaskCompletionSource<bool> _firstWriteStarted =
+            new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly string _faultMessage;
+        private int _writeCount;
+
+        /// <summary>Initializes a new instance of the FlushFaultingGatedStream class.</summary>
+        /// <param name="faultMessage">Message carried by the <see cref="IOException"/> every flush throws.</param>
+        public FlushFaultingGatedStream(string faultMessage)
+        {
+            _faultMessage = faultMessage;
+        }
+
+        /// <summary>Gets a task that completes once the first <see cref="WriteAsync"/> call has started blocking.</summary>
+        public Task FirstWriteStarted => _firstWriteStarted.Task;
+
+        /// <summary>Releases the first blocked write so it can complete.</summary>
+        public void ReleaseFirstWrite() => _release.Release();
+
         /// <inheritdoc />
         public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
@@ -910,8 +1175,7 @@ public sealed class WorkerFrameProtocolTests
         /// <inheritdoc />
         public override Task FlushAsync(CancellationToken cancellationToken)
         {
-            Interlocked.Increment(ref _flushCount);
-            return base.FlushAsync(cancellationToken);
+            return Task.FromException(new IOException(_faultMessage));
         }
 
         /// <inheritdoc />

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

@@ -912,6 +912,86 @@ public sealed class WorkerPipeSessionTests
         await Assert.ThrowsAsync<InvalidOperationException>(async () => await runTask);
     }
 
+    /// <summary>
+    ///     WRK-31, the other side of the invariant. The graceful path leaves the message loop
+    ///     through its <c>return</c> after the shutdown ack, with that iteration's read already
+    ///     awaited — so there is no abandoned read for teardown to account for. Pinning this keeps
+    ///     the new disposal-and-observe step a pure no-op on the path production takes every time a
+    ///     session closes normally: no "PipeRead" observation, and therefore no chance of paying
+    ///     the observation timeout on a healthy shutdown.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RunAsync_GracefulShutdown_LeavesNoPendingPipeReadToObserve()
+    {
+        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(10));
+        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+        FakeRuntimeSession runtime = new();
+        RecordingWorkerLogger logger = new();
+        WorkerPipeSession session = CreatePipeSession(
+            pipePair.WorkerStream,
+            runtime,
+            new WorkerPipeSessionOptions
+            {
+                HeartbeatInterval = TimeSpan.FromMilliseconds(100),
+                HeartbeatGrace = TimeSpan.FromSeconds(5),
+            },
+            logger);
+        Task runTask = session.RunAsync(cancellation.Token);
+        await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
+
+        // Reads the ack and bounds RunAsync's completion at 5s.
+        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
+
+        Assert.True(runtime.Disposed, "Graceful shutdown must dispose the runtime session.");
+        Assert.DoesNotContain(
+            logger.Events,
+            entry => (entry.EventName == "WorkerPipeSessionBackgroundTaskStopFailed"
+                    || entry.EventName == "WorkerPipeSessionBackgroundTaskStopTimedOut")
+                && entry.Fields.TryGetValue("task", out object? task)
+                && (task as string) == "PipeRead");
+    }
+
+    /// <summary>
+    ///     WRK-31. The session now owns and closes the transport rather than leaving it to
+    ///     <c>WorkerPipeClient</c>'s <c>using</c>, because the read it unblocks has to be observed
+    ///     while the session still holds it. This asserts the closure actually happens on the
+    ///     session's own timeline: the gateway end of the pipe must see disconnection while
+    ///     <see cref="PipePair"/> is still undisposed, so nothing but the worker side of the
+    ///     session can have closed it.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RunAsync_WhenSessionEnds_ClosesTransportBeforeTheHarnessDoes()
+    {
+        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(10));
+        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+        FakeRuntimeSession runtime = new();
+        WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime);
+        Task runTask = session.RunAsync(cancellation.Token);
+        await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
+
+        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
+
+        // PipePair.Dispose has not run — its `using` is still in scope — so the only thing that can
+        // have closed the worker end is the session. A gateway-side read is uncancellable on net48
+        // exactly as the worker's is, so a session that left the pipe open would park this read
+        // until the harness disposes; the bound below is what catches that.
+        Task<Exception> disconnectTask = ReadUntilDisconnectedAsync(pipePair.GatewayReader);
+        Task completedTask = await Task.WhenAny(
+            disconnectTask,
+            Task.Delay(TimeSpan.FromSeconds(5), cancellation.Token));
+        Assert.Same(disconnectTask, completedTask);
+
+        Exception disconnect = await disconnectTask;
+        Assert.True(
+            disconnect is IOException
+                || disconnect is ObjectDisposedException
+                || (disconnect is WorkerFrameProtocolException frameException
+                    && frameException.ErrorCode == WorkerFrameProtocolErrorCode.EndOfStream),
+            $"Expected the gateway read to observe a pipe disconnection, got {disconnect}.");
+    }
+
     /// <summary>
     ///     Verifies that ShutdownWorker returns its OK reply BEFORE the graceful
     ///     shutdown runs and disposes the runtime session, and that the message
@@ -1881,7 +1961,11 @@ public sealed class WorkerPipeSessionTests
             () => 1234,
             sessionOptions,
             () => runtime,
-            logger);
+            logger,
+            // Hand the session the same ownership the production WorkerPipeClient path gives it, so
+            // these tests exercise the real teardown: the session closes the transport itself and
+            // then observes the read that closure unblocks.
+            transportStream: stream);
     }
 
     private static WorkerFrameProtocolOptions CreateOptions()
@@ -2149,6 +2233,32 @@ public sealed class WorkerPipeSessionTests
         }
     }
 
+    /// <summary>
+    ///     Reads a pipe end until it stops producing frames, returning whatever ended it. Frames
+    ///     still buffered from before the peer closed its handle — a trailing heartbeat, say — are
+    ///     drained first, because Windows named pipes hand over buffered bytes ahead of the
+    ///     broken-pipe signal.
+    /// </summary>
+    /// <param name="reader">Frame reader over the end being watched.</param>
+    /// <returns>The exception that ended the read.</returns>
+    private static async Task<Exception> ReadUntilDisconnectedAsync(WorkerFrameReader reader)
+    {
+        while (true)
+        {
+            try
+            {
+                await reader.ReadAsync(CancellationToken.None).ConfigureAwait(false);
+            }
+            catch (Exception exception) when (
+                exception is IOException
+                || exception is ObjectDisposedException
+                || exception is WorkerFrameProtocolException)
+            {
+                return exception;
+            }
+        }
+    }
+
     private static WorkerEnvelope[] ReadWrittenFrames(
         MemoryStream stream,
         WorkerFrameProtocolOptions options)
@@ -2165,6 +2275,123 @@ public sealed class WorkerPipeSessionTests
         return envelopes.ToArray();
     }
 
+    /// <summary>
+    ///     The one teardown test that has to arm <see cref="TaskScheduler.UnobservedTaskException"/>,
+    ///     which is process-global: a task faulting in any concurrently running test class can be
+    ///     finalized inside this test's window and read as its result. It therefore lives in its own
+    ///     non-parallel collection (see <see cref="WorkerPipeSessionNonParallelCollection"/>) rather
+    ///     than alongside its siblings. Nested so it can still reach
+    ///     <see cref="WorkerPipeSessionTests"/>'s private harness — <c>PipePair</c>,
+    ///     <c>CreatePipeSession</c>, <c>RecordingWorkerLogger</c> — without widening any of it.
+    /// </summary>
+    [Collection(WorkerPipeSessionNonParallelCollection.Name)]
+    public sealed class AbandonedPipeReadTeardownTests
+    {
+        /// <summary>
+        ///     WRK-31. A fault exit unwinds the message loop while its frame read is still pending,
+        ///     and on net48 nothing can cancel that read — <c>NamedPipeClientStream.ReadAsync</c>
+        ///     ignores the token, so only closing the handle ends it. The session must therefore
+        ///     dispose the transport itself and account for the read that disposal unblocks: the
+        ///     worker installs no <c>TaskScheduler.UnobservedTaskException</c> handler, so before
+        ///     this the read faulted on a task nobody held — still carrying the reader's reused
+        ///     prefix buffer and its pooled payload buffer — and surfaced only at finalization.
+        /// </summary>
+        /// <returns>A task that represents the asynchronous operation.</returns>
+        [Fact]
+        public async Task RunAsync_WhenFaultEndsSession_ObservesTheAbandonedPipeRead()
+        {
+            const uint tinyMaxFrameBytes = 4096;
+            object unobservedGate = new();
+            List<Exception> unobservedPipeExceptions = new();
+            EventHandler<UnobservedTaskExceptionEventArgs> unobservedHandler = (_, args) =>
+            {
+                // Narrowed to a pipe stream's own teardown exception even though the collection is
+                // non-parallel, because the handler stays armed across this test's own async
+                // machinery. SetObserved is deliberately NOT called — the default policy already
+                // swallows these, and observing them here would mask a regression rather than
+                // report it.
+                foreach (Exception inner in args.Exception.Flatten().InnerExceptions)
+                {
+                    if ((inner is ObjectDisposedException || inner is IOException)
+                        && inner.Message.IndexOf("pipe", StringComparison.OrdinalIgnoreCase) >= 0)
+                    {
+                        lock (unobservedGate)
+                        {
+                            unobservedPipeExceptions.Add(inner);
+                        }
+                    }
+                }
+            };
+
+            RecordingWorkerLogger logger = new();
+            TaskScheduler.UnobservedTaskException += unobservedHandler;
+            try
+            {
+                using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15));
+                using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+                FakeRuntimeSession runtime = new();
+                WorkerPipeSession session = CreatePipeSession(
+                    pipePair.WorkerStream,
+                    runtime,
+                    new WorkerPipeSessionOptions
+                    {
+                        HeartbeatInterval = TimeSpan.FromMilliseconds(100),
+                        HeartbeatGrace = TimeSpan.FromSeconds(5),
+                    },
+                    logger);
+                runtime.EnqueueEvent(CreateOversizedWorkerEvent(sequence: 91, payloadBytes: 16 * 1024));
+                Task runTask = session.RunAsync(cancellation.Token);
+                await CompleteGatewayHandshakeAsync(pipePair, tinyMaxFrameBytes, cancellation.Token);
+
+                await ReadUntilAsync(
+                    pipePair.GatewayReader,
+                    WorkerEnvelope.BodyOneofCase.WorkerFault,
+                    cancellation.Token);
+
+                // The same 5s bound the sibling oversized-event test uses: teardown must not stall
+                // on the read it abandoned.
+                Task completedTask = await Task.WhenAny(
+                    runTask,
+                    Task.Delay(TimeSpan.FromSeconds(5), cancellation.Token));
+                Assert.Same(runTask, completedTask);
+                await Assert.ThrowsAsync<InvalidOperationException>(async () => await runTask);
+
+                // Evidence the read was abandoned and that teardown took responsibility for it: the
+                // shared observe-with-timeout helper records it under the "PipeRead" tag either way.
+                // Which of the two entries lands is a timing detail, not a contract. The fault
+                // normally arrives at once (StopFailed), but Windows owes no deadline for a
+                // completion torn off a closed handle, so on a loaded box it can arrive after
+                // BackgroundTaskStopTimeout (StopTimedOut). Both are correct, because observation is
+                // unconditional — the helper attaches a fault-observing continuation when it gives
+                // up waiting — and the unobserved-exception assertion below is what actually pins
+                // that. That the transport really is closed is pinned deterministically by
+                // RunAsync_WhenSessionEnds_ClosesTransportBeforeTheHarnessDoes, so insisting on
+                // "StopFailed within 1s" here would buy nothing but a flake at the windev gate.
+                Assert.Contains(
+                    logger.Events,
+                    entry => (entry.EventName == "WorkerPipeSessionBackgroundTaskStopFailed"
+                            || entry.EventName == "WorkerPipeSessionBackgroundTaskStopTimedOut")
+                        && entry.Fields.TryGetValue("task", out object? task)
+                        && (task as string) == "PipeRead");
+
+                // Drive any task that faulted without an awaiter through its finalizer, which is
+                // what raises UnobservedTaskException. Nothing from the pipe read may surface.
+                GC.Collect();
+                GC.WaitForPendingFinalizers();
+                GC.Collect();
+            }
+            finally
+            {
+                TaskScheduler.UnobservedTaskException -= unobservedHandler;
+            }
+
+            lock (unobservedGate)
+            {
+                Assert.Empty(unobservedPipeExceptions);
+            }
+        }
+    }
+
     private sealed class RecordingWorkerLogger : ZB.MOM.WW.MxGateway.Worker.Bootstrap.IWorkerLogger
     {
         private readonly object gate = new();
@@ -2484,3 +2711,19 @@ public sealed class WorkerPipeSessionTests
         }
     }
 }
+
+/// <summary>
+///     Collection for tests that observe process-global state and so cannot share the runner with
+///     anything else. Its only member today is
+///     <see cref="WorkerPipeSessionTests.AbandonedPipeReadTeardownTests"/>, which arms
+///     <see cref="TaskScheduler.UnobservedTaskException"/> and forces a GC: a task faulting in any
+///     concurrently running test class would be finalized inside that window and misread as this
+///     session's orphaned pipe read. Keep membership minimal — every test added here is a test the
+///     rest of the suite has to wait for.
+/// </summary>
+[CollectionDefinition(Name, DisableParallelization = true)]
+public sealed class WorkerPipeSessionNonParallelCollection
+{
+    /// <summary>Collection name referenced by <see cref="CollectionAttribute"/>.</summary>
+    public const string Name = "WorkerPipeSessionNonParallel";
+}

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

@@ -669,6 +669,107 @@ public sealed class MxAccessCommandExecutorTests
         Assert.Contains("RemoveItem:90:900", fakeComObject.OperationNames);
     }
 
+    /// <summary>
+    ///     Verifies ReadBulk's cached fast path end to end — the
+    ///     <c>was_cached = true</c> half of the command, which nothing else in
+    ///     the worker suite exercises. With the tag already added AND advised
+    ///     and a value in the per-session cache, the executor answers from the
+    ///     cache: the result is successful and flagged cached, and no COM call
+    ///     is made for the read at all, so the subscription the caller did not
+    ///     create is left exactly as it was.
+    ///
+    ///     It also pins the borrow contract through the whole read path: the
+    ///     <c>Value</c>, <c>SourceTimestamp</c>, and status row on the
+    ///     <c>BulkReadResult</c> are the cached event's own instances, not
+    ///     copies (see <see cref="MxAccessValueCache.Set"/>). The worker only
+    ///     serializes them onto the IPC pipe, so the alias never escapes the
+    ///     process.
+    ///
+    ///     Driven through <see cref="MxAccessCommandExecutor"/> directly rather
+    ///     than <see cref="MxAccessStaSession"/> because seeding the cache needs
+    ///     a handle on it: <see cref="MxAccessSession.Create"/> only shares the
+    ///     sink's cache for the production <c>MxAccessBaseEventSink</c>, which
+    ///     casts the COM object to <c>LMXProxyServerClass</c> and so cannot take
+    ///     a fake, while <c>CreateForTesting</c> accepts the cache directly. The
+    ///     cached path neither waits nor pumps, so it needs no STA.
+    /// </summary>
+    [Fact]
+    public void Execute_ReadBulk_WhenTagIsAdvisedAndCached_ServesTheCachedInstancesWithoutTouchingTheSubscription()
+    {
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 92,
+            addItemHandle: 920);
+        MxAccessValueCache valueCache = new();
+        using MxAccessSession session = MxAccessSession.CreateForTesting(
+            mxAccessServer: fakeComObject,
+            eventSink: new NoopEventSink(),
+            valueCache: valueCache);
+        MxAccessCommandExecutor executor = new(
+            session,
+            new ZB.MOM.WW.MxGateway.Worker.Conversion.VariantConverter());
+
+        // Registry half of the fast path: the tag must resolve to a live item
+        // handle on this server AND carry an advice, or TryGetCachedReadFor
+        // falls through to the AddItem/Advise snapshot lifecycle.
+        MxCommandReply registerReply = executor.Execute(
+            CreateRegisterCommand("register-before-cached-read", "client-a"));
+        MxCommandReply addItemReply = executor.Execute(
+            CreateAddItemCommand("add-before-cached-read", 92, "Galaxy.Tag.Value"));
+        MxCommandReply adviseReply = executor.Execute(
+            CreateAdviseCommand("advise-before-cached-read", 92, 920));
+        Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code);
+        Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code);
+        Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code);
+
+        // Cache half: stand in for the event sink's post-publish hook, which
+        // records the event it just enqueued.
+        MxEvent cachedEvent = new()
+        {
+            Family = MxEventFamily.OnDataChange,
+            ServerHandle = 92,
+            ItemHandle = 920,
+            Quality = 192,
+            SourceTimestamp = Timestamp.FromDateTime(new(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc)),
+            Value = new MxValue
+            {
+                DataType = MxDataType.Integer,
+                VariantType = "VT_I4",
+                Int32Value = 7788,
+            },
+            OnDataChange = new OnDataChangeEvent(),
+        };
+        cachedEvent.Statuses.Add(new MxStatusProxy { Category = MxStatusCategory.Ok });
+        valueCache.Set(92, 920, cachedEvent);
+
+        MxCommandReply reply = executor.Execute(CreateReadBulkCommand(
+            "read-bulk-cached",
+            serverHandle: 92,
+            tagAddresses: new[] { "Galaxy.Tag.Value" },
+            timeoutMs: 80));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Equal(MxCommandKind.ReadBulk, reply.Kind);
+        BulkReadResult result = Assert.Single(reply.ReadBulk.Results);
+        Assert.True(result.WasSuccessful);
+        Assert.True(result.WasCached);
+        Assert.Equal("Galaxy.Tag.Value", result.TagAddress);
+        Assert.Equal(920, result.ItemHandle);
+        Assert.Equal(192, result.Quality);
+        Assert.Equal(7788, result.Value.Int32Value);
+
+        // Borrowed, not copied — all the way from the event handed to
+        // MxAccessValueCache.Set out to the reply the worker serializes.
+        Assert.Same(cachedEvent.Value, result.Value);
+        Assert.Same(cachedEvent.SourceTimestamp, result.SourceTimestamp);
+        Assert.Same(cachedEvent.Statuses[0], Assert.Single(result.Statuses));
+
+        // No second AddItem, and above all no UnAdvise/RemoveItem: only the
+        // three setup calls ever reached MXAccess.
+        Assert.Equal(
+            new[] { "Register:client-a", "AddItem:92:Galaxy.Tag.Value", "Advise:92:920" },
+            fakeComObject.OperationNames);
+    }
+
     /// <summary>Verifies that ReadBulk with no payload returns an invalid request error.</summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
     [Fact]

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

@@ -48,14 +48,28 @@ public sealed class MxAccessValueCacheTests
     }
 
     /// <summary>
-    ///     Verifies that Set stores an independent deep-copied snapshot: mutating
-    ///     the source event's protobuf sub-messages after caching does not alter
-    ///     the cached value. WRK-11 stopped the event sink cloning before enqueue,
-    ///     so the same MxEvent instance now flows to the outbound queue; the cache
-    ///     must own its own copy so the two never share mutable state.
+    ///     Pins the ownership contract: <c>Set</c> borrows the event's own
+    ///     protobuf sub-messages instead of deep-copying them, so <c>TryGet</c>
+    ///     hands back the very <c>Value</c>, <c>SourceTimestamp</c>, and
+    ///     <c>MxStatusProxy</c> instances the caller passed in.
+    ///
+    ///     This is safe only because the event is write-once by the time
+    ///     <c>Set</c> runs: the sink enqueues it (which stamps the worker
+    ///     sequence and timestamp) and only then post-publishes it here, and
+    ///     <see cref="MxAccessEventQueue"/>'s ownership invariant forbids
+    ///     mutating an enqueued event. This test therefore asserts reference
+    ///     identity and deliberately does NOT mutate the event afterwards —
+    ///     doing so is exactly what the contract forbids, and it would also
+    ///     invalidate the serialized size the queue memoized at enqueue.
+    ///
+    ///     It replaced a test asserting the opposite (an independent deep-copied
+    ///     snapshot). The three clones that test pinned — the MxValue, the
+    ///     Timestamp, and the RepeatedField plus every status row in it — ran on
+    ///     every OnDataChange and bought nothing: the read path already aliased
+    ///     the cache's instances into every BulkReadResult.
     /// </summary>
     [Fact]
-    public void Set_StoresIndependentSnapshot_UnaffectedByLaterEventMutation()
+    public void Set_BorrowsTheEventsOwnInstances_ByOwnershipContract()
     {
         MxAccessValueCache cache = new();
         Timestamp sourceTimestamp = Timestamp.FromDateTime(new(2026, 5, 19, 9, 0, 0, DateTimeKind.Utc));
@@ -63,19 +77,12 @@ public sealed class MxAccessValueCacheTests
 
         cache.Set(7, 21, mxEvent);
 
-        // Mutate the event in place after it was cached — as if it kept flowing
-        // through the (unrelated) outbound path. None of this must reach the cache.
-        mxEvent.Value.Int32Value = 999;
-        mxEvent.Quality = 0;
-        mxEvent.SourceTimestamp = Timestamp.FromDateTime(new(2030, 1, 1, 0, 0, 0, DateTimeKind.Utc));
-        mxEvent.Statuses[0].Category = MxStatusCategory.SecurityError;
-
         Assert.True(cache.TryGet(7, 21, out MxAccessValueCache.CachedValue cached));
-        Assert.Equal(100, cached.Value.Int32Value);
+        Assert.Same(mxEvent.Value, cached.Value);
+        Assert.Same(mxEvent.SourceTimestamp, cached.SourceTimestamp);
+        Assert.Same(mxEvent.Statuses, cached.Statuses);
+        Assert.Same(mxEvent.Statuses[0], Assert.Single(cached.Statuses));
         Assert.Equal(192, cached.Quality);
-        Assert.Equal(sourceTimestamp, cached.SourceTimestamp);
-        Assert.Single(cached.Statuses);
-        Assert.Equal(MxStatusCategory.Ok, cached.Statuses[0].Category);
     }
 
     /// <summary>Verifies that TryGet returns false for unknown handles.</summary>

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

@@ -22,6 +22,14 @@ public sealed class WorkerFrameReader
 
     // Reused across frames rather than allocated per read (GWC-30). Safe because ReadAsync is
     // single-consumer by construction; the prefix is fully overwritten by every read.
+    //
+    // The single-consumer invariant has to survive teardown as well as steady state, because a read
+    // abandoned by WorkerPipeSession's message loop still owns this buffer (and its rented payload
+    // buffer) until it faults. Nothing may call ReadAsync again after that point: a second read
+    // would race the abandoned one for the prefix, and could hand a pooled payload buffer back to
+    // ArrayPool twice. The loop guarantees it structurally — it only ever issues a read after
+    // awaiting the previous one, and it never re-enters after unwinding — and teardown only awaits
+    // the abandoned read, never reissues it (WorkerPipeSession.ObserveAbandonedPipeReadAsync).
     private readonly byte[] _lengthPrefix = new byte[sizeof(uint)];
 
     /// <summary>Initializes the reader with a stream and protocol options.</summary>
@@ -106,6 +114,12 @@ public sealed class WorkerFrameReader
         int offset = 0;
         while (offset < count)
         {
+            // The token is forwarded but is NOT a bound on a pipe read: on .NET Framework 4.8
+            // NamedPipeClientStream.ReadAsync accepts a CancellationToken and never wires it to the
+            // overlapped I/O, so a read waiting on gateway bytes ignores cancellation entirely. Only
+            // closing the handle ends it (WorkerPipeSession disposes the transport at teardown for
+            // exactly this reason). It is still passed because non-pipe streams — the
+            // MemoryStream-backed unit tests, and any future transport — do honor it.
             int bytesRead = await _stream
                 .ReadAsync(buffer, offset, count - offset, cancellationToken)
                 .ConfigureAwait(false);

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

@@ -14,9 +14,11 @@ namespace ZB.MOM.WW.MxGateway.Worker.Ipc;
 /// Writes worker frames to a stream with length-prefixed protobuf serialization. Callers enqueue a
 /// frame at a <see cref="WorkerFrameWritePriority"/> and then contend for a single write lock; whoever
 /// holds the lock drains every queued frame, control frames first, so a reply, fault, or heartbeat is
-/// never delayed behind an event backlog. The envelope <c>Sequence</c> is stamped by the
-/// draining lock-holder at the moment of writing, so the on-wire order and the stamped sequence always
-/// agree even under concurrent callers and priority reordering.
+/// never delayed behind an event backlog — neither in the bytes it writes nor in the flush that
+/// delivers them, because the drain flushes at every control-to-event boundary rather than only at the
+/// end of the pass. The envelope <c>Sequence</c> is stamped by the draining lock-holder at the moment
+/// of writing, so the on-wire order and the stamped sequence always agree even under concurrent callers
+/// and priority reordering.
 /// </summary>
 public sealed class WorkerFrameWriter
 {
@@ -24,15 +26,25 @@ public sealed class WorkerFrameWriter
     {
         /// <summary>Initializes a new instance of the PendingFrame class.</summary>
         /// <param name="envelope">Worker envelope awaiting write.</param>
-        public PendingFrame(WorkerEnvelope envelope)
+        /// <param name="priority">Priority class the frame was queued at.</param>
+        public PendingFrame(WorkerEnvelope envelope, WorkerFrameWritePriority priority)
         {
             Envelope = envelope;
+            IsControl = priority != WorkerFrameWritePriority.Event;
             Completion = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
         }
 
         /// <summary>Gets the worker envelope awaiting write.</summary>
         public WorkerEnvelope Envelope { get; }
 
+        /// <summary>
+        /// Gets a value indicating whether this frame was queued as control-plane traffic. Recorded at
+        /// construction from the same expression that picks the queue, so the class the drain sees can
+        /// never disagree with the queue the frame sits in. The drain uses it to flush and complete
+        /// written control frames at the moment it turns to events (see <see cref="DrainQueuedFramesAsync"/>).
+        /// </summary>
+        public bool IsControl { get; }
+
         /// <summary>Gets the completion source signaled once the frame has been written or has failed.</summary>
         public TaskCompletionSource<bool> Completion { get; }
 
@@ -95,6 +107,15 @@ public sealed class WorkerFrameWriter
     /// the canceller behind the very write it is abandoning would defeat the point of cancellation.
     /// The abandoned frame's completion gets a fault-observing continuation so a write failure after
     /// the caller unwinds never raises an unobserved-task exception (NEXT-04).
+    /// <para>
+    /// Latency contract: a control frame's bytes are written, flushed, and its completion
+    /// resolved before the events a drain pass writes after it — the delivery point of a heartbeat,
+    /// reply, fault, or shutdown ack is never charged for the event backlog behind it. The returned
+    /// task can still be later than that instant for a caller that lost the write-lock race: it only
+    /// observes its completion after the winning drainer releases the lock, so its own return remains
+    /// bounded by that pass. That parking is deliberate — the alternative is to race the lock wait
+    /// against the completion, which buys nothing for the frame's delivery.
+    /// </para>
     /// </remarks>
     public async Task WriteAsync(
         WorkerEnvelope envelope,
@@ -106,7 +127,7 @@ public sealed class WorkerFrameWriter
             throw new ArgumentNullException(nameof(envelope));
         }
 
-        PendingFrame frame = new PendingFrame(envelope);
+        PendingFrame frame = new PendingFrame(envelope, priority);
         lock (_gate)
         {
             if (priority == WorkerFrameWritePriority.Event)
@@ -153,8 +174,12 @@ public sealed class WorkerFrameWriter
     /// rather than one per frame (WRK-25, realizing the WRK-12 coalescing on the path it was built
     /// for). Intra-batch order is preserved because the enqueue is atomic under <c>_gate</c> and each
     /// class queue is FIFO; the control-before-event guarantee still holds because any concurrently
-    /// queued control frame is drained ahead of this batch by <see cref="DequeueNext"/>. Every frame's
-    /// "written and flushed before completion" contract is unchanged.
+    /// queued control frame is drained ahead of this batch by <see cref="DequeueNext"/>, and — since
+    /// the control-frame completion decoupling — is also flushed and completed before this batch's
+    /// remaining events are written, so a batch in flight does not delay a control frame's delivery.
+    /// Every frame's "written and flushed before completion" contract is unchanged. An event batch that
+    /// a control frame cuts into therefore pays one extra flush; an uninterrupted batch still pays
+    /// exactly one.
     /// </summary>
     /// <param name="envelopes">Envelopes to write, in order.</param>
     /// <param name="priority">Scheduling priority for the whole batch.</param>
@@ -189,7 +214,7 @@ public sealed class WorkerFrameWriter
         {
             WorkerEnvelope envelope = envelopes[index]
                 ?? throw new ArgumentException("Batch envelopes must not contain null.", nameof(envelopes));
-            frames[index] = new PendingFrame(envelope);
+            frames[index] = new PendingFrame(envelope, priority);
         }
 
         lock (_gate)
@@ -296,14 +321,24 @@ public sealed class WorkerFrameWriter
     // The stream write itself is not cancellable: a frame is written atomically or fails, never left
     // half-written on the pipe because a caller gave up waiting.
     //
-    // Flushes are coalesced across the whole drained batch (WRK-12 / IPC-15): each frame is written to
-    // the stream but not flushed individually; a single FlushAsync runs after the batch, then every
-    // successfully-written frame is completed. A caller's Completion therefore still signals only after
-    // its bytes have been written AND flushed, so the "written and flushed" contract is unchanged — but
-    // a burst of N events now costs one flush syscall instead of N.
+    // Flushes are coalesced within a priority class rather than blindly across the whole pass (WRK-12 /
+    // IPC-15, narrowed by the control-frame completion decoupling): each frame is written to the stream
+    // but not flushed individually, and one FlushAsync runs at the end of the pass — plus one at each
+    // control-to-event boundary, which flushes and completes the control frames written so far before
+    // the event backlog behind them is written, instead of after it. Without that boundary flush the
+    // priority scheduler only got control *bytes* out early: their delivery point, and every waiting
+    // caller's completion, still sat behind up to a full event batch.
+    //
+    // A caller's Completion therefore still signals only after its bytes have been written AND flushed —
+    // the contract is unchanged, the moment it is reached simply stops being pinned to the end of the
+    // pass. Cost is bounded: a pure-event pass (the event hot path) still pays exactly one flush, a burst
+    // of control frames still pays one for the whole burst, and only a pass that actually mixes both
+    // classes pays a second — never one flush per control frame, which is the syscall-per-heartbeat cost
+    // WRK-12 removed.
     private async Task DrainQueuedFramesAsync()
     {
         List<PendingFrame> written = new List<PendingFrame>();
+        bool writtenHoldsControl = false;
         while (true)
         {
             PendingFrame? frame = DequeueNext();
@@ -312,10 +347,40 @@ public sealed class WorkerFrameWriter
                 break;
             }
 
+            if (writtenHoldsControl && !frame.IsControl)
+            {
+                // Class transition: the frames written so far include at least one control frame whose
+                // caller is waiting on delivery. Flush and complete them here rather than parking them
+                // behind the events this pass is about to write. Charged once per transition, not once
+                // per control frame. Event frames already in the list ride along — they too are written
+                // and now flushed, so completing them early is the same contract, earlier.
+                try
+                {
+                    await _stream.FlushAsync(CancellationToken.None).ConfigureAwait(false);
+                }
+                catch (Exception exception)
+                {
+                    // Same shape as the end-of-pass flush failure: the bytes reached the stream but the
+                    // flush that guarantees delivery failed, so the pipe is broken. Fail the frame just
+                    // claimed (it is out of its queue and nothing else will ever complete it), every
+                    // written-but-unflushed frame, and everything still queued, then stop draining.
+                    frame.Completion.TrySetException(exception);
+                    FailFrames(written, exception);
+                    FailAllQueued(exception);
+                    return;
+                }
+
+                // Completed frames leave the list, so a later failure in this pass cannot fail them.
+                CompleteFrames(written);
+                written.Clear();
+                writtenHoldsControl = false;
+            }
+
             try
             {
                 await WriteFrameAsync(frame.Envelope).ConfigureAwait(false);
                 written.Add(frame);
+                writtenHoldsControl |= frame.IsControl;
             }
             catch (WorkerFrameProtocolException exception) when (IsPerFrameRejection(exception))
             {
@@ -348,13 +413,19 @@ public sealed class WorkerFrameWriter
         catch (Exception exception)
         {
             // The batch reached the stream but the flush that guarantees delivery failed: the pipe is
-            // broken. Fail every frame in the batch (the queue was already drained) so no caller treats
-            // an unflushed write as delivered.
+            // broken. Fail every frame still in the batch (the queue was already drained) so no caller
+            // treats an unflushed write as delivered. Frames a boundary flush already completed are not
+            // in the list — their bytes were flushed, so this later failure does not reach back to them.
             FailFrames(written, exception);
             return;
         }
 
-        foreach (PendingFrame frame in written)
+        CompleteFrames(written);
+    }
+
+    private static void CompleteFrames(List<PendingFrame> frames)
+    {
+        foreach (PendingFrame frame in frames)
         {
             frame.Completion.TrySetResult(true);
         }

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

@@ -151,6 +151,13 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
 
         WorkerFrameProtocolOptions frameOptions = new(options);
 
+        // The session disposes this pipe itself as its last teardown step — that disposal is what
+        // unblocks a net48 pipe read the message loop abandoned, and it has to happen while the
+        // session still holds the read task so the resulting fault is observed rather than orphaned
+        // (see WorkerPipeSession.RunAsync). The `using` stays as the backstop for the paths the
+        // session never reaches: a session factory that throws, or a RunAsync that never gets past
+        // its own construction. Disposal is idempotent, so the second Dispose is a no-op — do not
+        // "clean up" this `using` on the assumption that it is now redundant.
         using NamedPipeClientStream pipe = await ConnectWithRetryAsync(options.PipeName, cancellationToken)
             .ConfigureAwait(false);
 

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

@@ -39,8 +39,21 @@ public sealed class WorkerPipeSession
     private readonly WorkerFrameWriter _writer;
     private readonly object _commandTaskGate = new();
     private readonly HashSet<Task> _activeCommandTasks = new();
+
+    // The transport the reader and writer share, when this session was handed one to own. Null for
+    // the reader/writer constructors, whose callers keep ownership of whatever streams they built the
+    // pair over. Owning it is what lets teardown close the handle (WRK-31): on net48 a pipe read
+    // parked in the kernel cannot be cancelled, only unblocked by disposal.
+    private readonly Stream? _transportStream;
+
     private IWorkerRuntimeSession? _runtimeSession;
 
+    // The one outstanding, not-yet-awaited frame read, or null when no read is in flight. Written
+    // only by the message loop (which sets it at each read issue and clears it before awaiting that
+    // read itself) and read only by RunAsync's finally — which runs after the loop's task has
+    // completed, so the await supplies the happens-before edge and no interlock is needed.
+    private Task<WorkerEnvelope>? _pendingReadTask;
+
     // Mutated from the message loop, command tasks, the heartbeat loop and the
     // shutdown path; volatile so cross-thread reads observe the latest state
     // without tearing (WorkerState is an int-backed protobuf enum).
@@ -63,7 +76,8 @@ public sealed class WorkerPipeSession
             () => Process.GetCurrentProcess().Id,
             new WorkerPipeSessionOptions(),
             () => new MxAccessStaSession((eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null)),
-            logger)
+            logger,
+            stream)
     {
     }
 
@@ -96,6 +110,12 @@ public sealed class WorkerPipeSession
     /// <param name="sessionOptions">Session-specific options.</param>
     /// <param name="runtimeSessionFactory">Factory creating the MXAccess runtime session.</param>
     /// <param name="logger">Optional logger for diagnostic output.</param>
+    /// <param name="transportStream">
+    ///     Stream the reader and writer share, when this session is to own it. Supplying it makes
+    ///     <see cref="RunAsync"/> dispose the transport as its last teardown step, which is the only
+    ///     way to unblock a pending net48 pipe read (see <see cref="RunMessageLoopAsync"/>). Null
+    ///     leaves ownership — and the disposal — with the caller.
+    /// </param>
     public WorkerPipeSession(
         WorkerFrameReader reader,
         WorkerFrameWriter writer,
@@ -103,7 +123,8 @@ public sealed class WorkerPipeSession
         Func<int> processIdProvider,
         WorkerPipeSessionOptions sessionOptions,
         Func<IWorkerRuntimeSession> runtimeSessionFactory,
-        IWorkerLogger? logger = null)
+        IWorkerLogger? logger = null,
+        Stream? transportStream = null)
     {
         _reader = reader ?? throw new ArgumentNullException(nameof(reader));
         _writer = writer ?? throw new ArgumentNullException(nameof(writer));
@@ -112,6 +133,7 @@ public sealed class WorkerPipeSession
         _sessionOptions = sessionOptions ?? throw new ArgumentNullException(nameof(sessionOptions));
         _runtimeSessionFactory = runtimeSessionFactory ?? throw new ArgumentNullException(nameof(runtimeSessionFactory));
         _logger = logger;
+        _transportStream = transportStream;
         _sessionOptions.Validate();
     }
 
@@ -142,9 +164,117 @@ public sealed class WorkerPipeSession
             _runtimeSession?.Dispose();
             _runtimeSession = null;
             _state = WorkerState.Stopped;
+
+            // Closing the transport is what actually ends a pipe read parked in the kernel: on net48
+            // NamedPipeClientStream.ReadAsync ignores its CancellationToken, so the message loop's
+            // cancellation can never reach one (WRK-31). It is deliberately the LAST teardown step,
+            // because in the ordinary case every frame this session will ever write has completed by
+            // the time control reaches here: WorkerFrameWriter.WriteAsync signals only after the
+            // frame is written AND flushed, and each exit path awaits its final write before
+            // unwinding — the shutdown ack and shutdown-timeout fault inside the loop's dispatch, the
+            // event-drain and oversized-event faults inside the drain task the loop awaits, the
+            // watchdog fault inside the heartbeat task the loop awaits, and the handshake fault
+            // inside CompleteStartupHandshakeAsync's catch.
+            //
+            // "Ordinary" is the honest word, not "always": the loop's wait on the heartbeat and
+            // drain tasks is budgeted (BackgroundTaskStopTimeout), and a stream write is genuinely
+            // uncancellable, so a write that overran the budget can still be in flight against the
+            // stream being disposed here. In-flight command replies are in the same position, and
+            // they raced the identical disposal before this change (WorkerPipeClient's `using` fired
+            // on the very next statement after RunAsync). That is precisely why disposal below is
+            // exception-tolerant and why every abandoned task gets a fault-observing continuation
+            // from ObserveBackgroundTaskStopAsync — a write losing its stream mid-flight must be a
+            // logged non-event, not a lost terminal exception or an unobserved task.
+            //
+            // Owning the disposal here — rather than leaving it to WorkerPipeClient's `using` — is
+            // what makes the abandoned read observable: the fault it takes on disposal lands on a
+            // task this session still holds, so ObserveAbandonedPipeReadAsync can await it instead of
+            // leaving it for a TaskScheduler.UnobservedTaskException handler the worker does not have.
+            DisposeTransportStream();
+            await ObserveAbandonedPipeReadAsync().ConfigureAwait(false);
         }
     }
 
+    /// <summary>
+    ///     Disposes the transport this session owns, if it was handed one. Dispose-time failures are
+    ///     logged and swallowed: this runs inside <see cref="RunAsync"/>'s finally, where letting a
+    ///     failure escape would replace the exception that actually ended the session (a shutdown
+    ///     timeout, a protocol violation, an event too large to frame) with a far less actionable
+    ///     one. Disposal is idempotent, so <c>WorkerPipeClient</c>'s outer <c>using</c> re-disposing
+    ///     the same stream immediately afterwards is a no-op.
+    /// </summary>
+    /// <remarks>
+    ///     The catch is deliberately total rather than the <see cref="IOException"/> /
+    ///     <see cref="ObjectDisposedException"/> pair the fault-write paths use, matching
+    ///     <see cref="ObserveBackgroundTaskStopAsync"/>'s shape. Narrowing it to the expected types
+    ///     would let an unexpected one — a <c>Win32Exception</c> surfaced by the handle close, say —
+    ///     do the exact harm this guard exists to prevent. The rule is about the position in the
+    ///     code, not about which exceptions are plausible: nothing thrown while releasing a handle
+    ///     is more actionable than the session's terminal exception, so nothing may displace it.
+    /// </remarks>
+    private void DisposeTransportStream()
+    {
+        if (_transportStream is null)
+        {
+            return;
+        }
+
+        try
+        {
+            _transportStream.Dispose();
+        }
+        catch (Exception exception)
+        {
+            _logger?.Error(
+                "WorkerPipeSessionTransportDisposeFailed",
+                new Dictionary<string, object?>
+                {
+                    ["session_id"] = _options.SessionId,
+                    ["exception"] = exception.ToString(),
+                });
+        }
+    }
+
+    /// <summary>
+    ///     Awaits the frame read the message loop walked away from, if there is one.
+    ///     <see cref="RunMessageLoopAsync"/> races an uncancellable read against the heartbeat and
+    ///     event-drain loops, so every fault exit (event-drain fault, oversized event, heartbeat
+    ///     write failure) leaves a read outstanding on a task the loop never awaits again. Once
+    ///     <see cref="DisposeTransportStream"/> has closed the handle that read faults with
+    ///     <see cref="ObjectDisposedException"/>, <see cref="IOException"/>, or a zero-byte read
+    ///     mapped to <c>EndOfStream</c>, and that fault has to be observed — the worker installs no
+    ///     <c>TaskScheduler.UnobservedTaskException</c> handler.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         The fault is observed <em>unconditionally</em>; it is only <em>logged</em> within
+    ///         <see cref="BackgroundTaskStopTimeout"/>. Windows is under no obligation to deliver
+    ///         the abandoned read's completion inside that budget, so a bounded await alone would
+    ///         reopen the orphaning window it was added to close. <see cref="ObserveBackgroundTaskStopAsync"/>
+    ///         therefore hands the task a fault-observing continuation when it gives up waiting,
+    ///         which makes the budget a diagnostics decision rather than a correctness one.
+    ///     </para>
+    ///     <para>
+    ///         No second read can follow this one. The message loop only ever issues a read after
+    ///         awaiting the previous one, and it never re-enters after unwinding, so
+    ///         <see cref="WorkerFrameReader"/>'s single-consumer invariant — and with it the safety
+    ///         of its reused length-prefix buffer and its pooled payload buffer, which the abandoned
+    ///         read still owns until it faults — holds through teardown.
+    ///     </para>
+    /// </remarks>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    private async Task ObserveAbandonedPipeReadAsync()
+    {
+        Task<WorkerEnvelope>? readTask = _pendingReadTask;
+        _pendingReadTask = null;
+        if (readTask is null)
+        {
+            return;
+        }
+
+        await ObserveBackgroundTaskStopAsync(readTask, "PipeRead").ConfigureAwait(false);
+    }
+
     /// <summary>Completes the gateway startup handshake using default MXAccess initialization.</summary>
     /// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
     /// <returns>A task that represents the asynchronous operation.</returns>
@@ -264,6 +394,34 @@ public sealed class WorkerPipeSession
         return _writer.WriteAsync(CreateEnvelope(ready), cancellationToken);
     }
 
+    /// <summary>
+    ///     Runs the post-handshake message loop, racing one outstanding frame read against the
+    ///     heartbeat and event-drain loops.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         <c>loopCancellation</c> does NOT bound the read. On .NET Framework 4.8
+    ///         <c>NamedPipeClientStream.ReadAsync</c> accepts a <see cref="CancellationToken"/> and
+    ///         then ignores it — the token never reaches the overlapped I/O, so a read parked
+    ///         waiting for gateway bytes stays parked no matter what is cancelled. The token is
+    ///         still passed because the reader's contract takes one and a non-pipe stream (the
+    ///         MemoryStream-backed unit tests) does honor it. What actually ends a parked read is
+    ///         closing the handle, which <see cref="RunAsync"/> does at the end of teardown.
+    ///     </para>
+    ///     <para>
+    ///         That asymmetry is why the loop records its outstanding read in
+    ///         <c>_pendingReadTask</c>. Every fault exit — an event-drain fault, an event too large
+    ///         to frame, a failed heartbeat write — unwinds through <c>Task.WhenAny</c> while the
+    ///         read is still pending, and the fault that read eventually takes has to be observed by
+    ///         somebody (see <see cref="ObserveAbandonedPipeReadAsync"/>). The field is cleared
+    ///         before the loop awaits a read itself, so it is non-null exactly when a read is
+    ///         outstanding and unobserved. Graceful exits — the <c>return</c> below, after a
+    ///         <c>WorkerShutdown</c> envelope or a <c>ShutdownWorker</c> command — leave no pending
+    ///         read at all, so teardown's observation is a no-op there.
+    ///     </para>
+    /// </remarks>
+    /// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
     private async Task RunMessageLoopAsync(CancellationToken cancellationToken)
     {
         using CancellationTokenSource loopCancellation = CancellationTokenSource
@@ -272,7 +430,11 @@ public sealed class WorkerPipeSession
             .CreateLinkedTokenSource(cancellationToken);
         Task heartbeatTask = RunHeartbeatLoopAsync(heartbeatCancellation.Token);
         Task eventDrainTask = RunEventDrainLoopAsync(heartbeatCancellation.Token);
+        Debug.Assert(
+            _pendingReadTask is null,
+            "A frame read must never be issued while another is outstanding: WorkerFrameReader is single-consumer, and a second read would race the abandoned one for the reused prefix buffer and could return a pooled payload buffer twice.");
         Task<WorkerEnvelope> readTask = _reader.ReadAsync(loopCancellation.Token);
+        _pendingReadTask = readTask;
 
         try
         {
@@ -281,6 +443,9 @@ public sealed class WorkerPipeSession
                 Task completedTask = await Task.WhenAny(readTask, heartbeatTask, eventDrainTask).ConfigureAwait(false);
                 if (completedTask == readTask)
                 {
+                    // The loop observes this read itself, whether it yields an envelope or throws,
+                    // so it is no longer the abandoned one teardown has to account for.
+                    _pendingReadTask = null;
                     WorkerEnvelope envelope = await readTask.ConfigureAwait(false);
                     bool keepReading = await DispatchGatewayEnvelopeAsync(envelope, cancellationToken).ConfigureAwait(false);
                     if (!keepReading)
@@ -288,7 +453,11 @@ public sealed class WorkerPipeSession
                         return;
                     }
 
+                    Debug.Assert(
+                        _pendingReadTask is null,
+                        "The previous read must have been awaited before the next is issued: WorkerFrameReader is single-consumer.");
                     readTask = _reader.ReadAsync(loopCancellation.Token);
+                    _pendingReadTask = readTask;
                 }
                 else if (completedTask == heartbeatTask)
                 {
@@ -309,6 +478,32 @@ public sealed class WorkerPipeSession
         }
     }
 
+    /// <summary>
+    ///     Waits a bounded time for a background task to stop and records what happened.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         The budget bounds the <em>logging</em>, never the observation. Every task this is
+    ///         asked to observe is uncancellable at the point that matters — a net48 pipe read
+    ///         ignores its token outright, and <c>WorkerFrameWriter.WriteFrameAsync</c> issues the
+    ///         stream write under <c>CancellationToken.None</c> so a frame is never left
+    ///         half-written on the wire — so any of them can outlive the budget and only then fault,
+    ///         typically against a transport <see cref="RunAsync"/> has since disposed. Overrunning
+    ///         the budget and returning is therefore not enough: the task would be left with nobody
+    ///         holding it, which is the exact orphaning this method exists to prevent.
+    ///     </para>
+    ///     <para>
+    ///         So the timeout path hands the task a fault-observing continuation before returning.
+    ///         The fault is then observed unconditionally, whenever it arrives; the budget only
+    ///         decides whether it also gets logged here or is swallowed silently by the
+    ///         continuation. That distinction matters because the worker installs no
+    ///         <c>TaskScheduler.UnobservedTaskException</c> handler, so an unheld faulted task
+    ///         surfaces only at finalization.
+    ///     </para>
+    /// </remarks>
+    /// <param name="task">Background task being stopped.</param>
+    /// <param name="taskName">Name recorded in the diagnostic logs.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
     private async Task ObserveBackgroundTaskStopAsync(
         Task task,
         string taskName)
@@ -318,6 +513,7 @@ public sealed class WorkerPipeSession
             .ConfigureAwait(false);
         if (completedTask != task)
         {
+            ObserveFaultWhenever(task);
             _logger?.Error(
                 "WorkerPipeSessionBackgroundTaskStopTimedOut",
                 new Dictionary<string, object?>
@@ -347,6 +543,24 @@ public sealed class WorkerPipeSession
         }
     }
 
+    /// <summary>
+    ///     Attaches a continuation that observes <paramref name="task"/>'s exception whenever the
+    ///     task eventually faults, so a task nobody is awaiting any more can never reach the
+    ///     finalizer with an unobserved exception. Mirrors the shape
+    ///     <c>WorkerFrameWriter.ObserveAbandonedFault</c> uses for frames a cancelled caller stops
+    ///     awaiting (NEXT-04). Faulting is the only outcome that runs the continuation, and the
+    ///     continuation is scheduled inline, so this costs nothing on the ordinary path.
+    /// </summary>
+    /// <param name="task">Task that may fault after its awaiter has walked away.</param>
+    private static void ObserveFaultWhenever(Task task)
+    {
+        _ = task.ContinueWith(
+            static faultedTask => _ = faultedTask.Exception,
+            CancellationToken.None,
+            TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
+            TaskScheduler.Default);
+    }
+
     private async Task RunEventDrainLoopAsync(CancellationToken cancellationToken)
     {
         while (!cancellationToken.IsCancellationRequested)

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

@@ -20,8 +20,11 @@ namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
 ///     <see cref="MxEvent"/> that it does not retain, reuse, or mutate after the
 ///     call returns. All production callers (MxAccessBaseEventSink,
 ///     MxAccessAlarmEventSink, AlarmCommandHandler) build a new event per
-///     Enqueue via the mapper and satisfy this; the value cache stores its own
-///     independent snapshot (see <see cref="MxAccessValueCache.Set"/>).
+///     Enqueue via the mapper and satisfy this. "Does not mutate" is the load-
+///     bearing half for the post-publish value cache, which deliberately borrows
+///     the enqueued event's own value/timestamp/status instances rather than
+///     copying them (see <see cref="MxAccessValueCache.Set"/>) — it reads and
+///     serializes them, and this invariant is what keeps that safe.
 ///     <para>
 ///         The byte-budgeted <see cref="Drain(uint, int)"/> relies on the same invariant: each
 ///         event's serialized size is measured once at enqueue and stored beside it, which is

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

@@ -57,7 +57,12 @@ public sealed class MxAccessValueCache
     /// <summary>Records a fresh OnDataChange payload for the given handle pair.</summary>
     /// <param name="serverHandle">MXAccess server handle.</param>
     /// <param name="itemHandle">MXAccess item handle.</param>
-    /// <param name="mxEvent">The protobuf MxEvent created by the event mapper.</param>
+    /// <param name="mxEvent">
+    ///     The protobuf MxEvent created by the event mapper, already handed to
+    ///     the outbound queue. The cache borrows this event's
+    ///     <c>Value</c>/<c>SourceTimestamp</c>/<c>Statuses</c> instances rather
+    ///     than copying them — see the ownership contract in the method body.
+    /// </param>
     public void Set(
         int serverHandle,
         int itemHandle,
@@ -68,20 +73,39 @@ public sealed class MxAccessValueCache
             throw new ArgumentNullException(nameof(mxEvent));
         }
 
-        // WRK-11: the event sink no longer clones before enqueue, so the passed
-        // mxEvent is the very instance handed to the outbound queue. Deep-copy
-        // the value/timestamp/statuses payload we retain here so the cache's
-        // snapshot stays independent of the enqueued (and later serialized)
-        // event — the two must never share mutable protobuf sub-messages.
-        // Value is always set for OnDataChange; SourceTimestamp may be unset when
-        // the source timestamp could not be parsed, so both are cloned only when
-        // present. The null-forgiving result matches CachedValue's non-null-
-        // annotated parameters, which already accepted a runtime-null value or
-        // timestamp before WRK-11 (the ternary keeps the compiler's null-state
-        // from poisoning to maybe-null, which a plain null check would do).
-        MxValue cachedValue = mxEvent.Value is null ? null! : mxEvent.Value.Clone();
-        Timestamp cachedTimestamp = mxEvent.SourceTimestamp is null ? null! : mxEvent.SourceTimestamp.Clone();
-
+        // Ownership contract: the cache BORROWS, it does not copy. CachedValue
+        // retains the event's own Value, SourceTimestamp, and Statuses
+        // instances.
+        //
+        // Sound because the event is write-once by the time this runs.
+        // MxAccessBaseEventSink.EnqueueEvent calls eventQueue.Enqueue first —
+        // which stamps WorkerSequence/WorkerTimestamp inside the queue lock —
+        // and only then runs the postPublish hook that lands here; the queue's
+        // ownership invariant (see MxAccessEventQueue's class remarks) forbids
+        // mutating an event after it is enqueued. The producer side never
+        // reuses instances either: the mapper builds a fresh MxEvent, and the
+        // VariantConverter a fresh MxValue, per COM callback. The three deep
+        // copies this replaced (the MxValue — recursive for an MxArray — the
+        // Timestamp, and the RepeatedField container plus every MxStatusProxy
+        // in it) therefore bought nothing but garbage on the worker's hottest
+        // path.
+        //
+        // Consumers of TryGet / TryWaitForUpdate may read and serialize what
+        // they get back; they must never mutate it. ReadBulk already depends on
+        // that: MxAccessSession.SucceededRead puts these very instances on the
+        // BulkReadResult it returns, which the worker only serializes onto the
+        // IPC pipe — worker↔gateway is a process boundary, so no gateway-side
+        // consumer can alias them. Mutating one would corrupt the event still
+        // queued for the outbound stream AND invalidate QueuedEvent.Size, the
+        // serialized size memoized at enqueue that the byte-budgeted Drain
+        // charges against its budget: a message grown after enqueue could
+        // overshoot the negotiated frame max and fault the session with
+        // MessageTooLarge (WorkerPipeSession.FaultOnOversizedEventAsync).
+        //
+        // Value is always set for OnDataChange; SourceTimestamp can be null when
+        // the source timestamp could not be parsed. CachedValue's parameters are
+        // annotated non-null but have always accepted a runtime null for both,
+        // and the read side null-checks accordingly.
         long key = CreateItemKey(serverHandle, itemHandle);
         lock (syncRoot)
         {
@@ -91,10 +115,10 @@ public sealed class MxAccessValueCache
 
             entries[key] = new CachedValue(
                 nextVersion,
-                cachedValue,
+                mxEvent.Value,
                 mxEvent.Quality,
-                cachedTimestamp,
-                mxEvent.Statuses.Clone());
+                mxEvent.SourceTimestamp,
+                mxEvent.Statuses);
         }
 
         // Signaled outside the lock: the waiter re-takes syncRoot (through
@@ -228,23 +252,35 @@ public sealed class MxAccessValueCache
     }
 
     /// <summary>
-    ///     Snapshot of the most recent OnDataChange payload for a handle pair.
+    ///     The most recent OnDataChange payload for a handle pair.
     ///     <see cref="Version"/> increments by one on every <see cref="Set"/>
     ///     call so the bulk read executor can detect "a new value arrived
     ///     since I started waiting".
     /// </summary>
     /// <remarks>
-    ///     Plain readonly struct (not a record) so this compiles under the
-    ///     worker's net48 target, which lacks <c>IsExternalInit</c>.
+    ///     <para>
+    ///         Borrowed references, not copies. <see cref="Value"/>,
+    ///         <see cref="SourceTimestamp"/>, and <see cref="Statuses"/> are the
+    ///         very instances hanging off the MxEvent that was enqueued for the
+    ///         outbound stream — <see cref="Set"/> carries the full ownership
+    ///         contract. Read them and serialize them; never mutate them and
+    ///         never hand them to something that will. That event is write-once
+    ///         from the moment it is enqueued, and its serialized size is
+    ///         memoized at that point for the byte-budgeted drain.
+    ///     </para>
+    ///     <para>
+    ///         Plain readonly struct (not a record) so this compiles under the
+    ///         worker's net48 target, which lacks <c>IsExternalInit</c>.
+    ///     </para>
     /// </remarks>
     public readonly struct CachedValue
     {
         /// <summary>Initializes a new cached value snapshot.</summary>
         /// <param name="version">Version counter incremented on each update.</param>
-        /// <param name="value">The MXAccess value.</param>
+        /// <param name="value">The MXAccess value, borrowed from the enqueued event.</param>
         /// <param name="quality">The MXAccess quality code.</param>
-        /// <param name="sourceTimestamp">The source timestamp of the value.</param>
-        /// <param name="statuses">The MXAccess status codes.</param>
+        /// <param name="sourceTimestamp">The source timestamp of the value, borrowed from the enqueued event.</param>
+        /// <param name="statuses">The MXAccess status codes, borrowed from the enqueued event.</param>
         public CachedValue(
             ulong version,
             MxValue value,
@@ -262,16 +298,16 @@ public sealed class MxAccessValueCache
         /// <summary>Monotonic per-handle version counter.</summary>
         public ulong Version { get; }
 
-        /// <summary>The cached MxValue payload.</summary>
+        /// <summary>The OnDataChange event's own MxValue payload. Read-only to consumers.</summary>
         public MxValue Value { get; }
 
         /// <summary>Quality code from the OnDataChange event.</summary>
         public int Quality { get; }
 
-        /// <summary>Source timestamp from the OnDataChange event.</summary>
+        /// <summary>The OnDataChange event's own source timestamp. Read-only to consumers.</summary>
         public Timestamp SourceTimestamp { get; }
 
-        /// <summary>MxStatusProxy entries from the OnDataChange event.</summary>
+        /// <summary>The OnDataChange event's own MxStatusProxy collection. Read-only to consumers.</summary>
         public RepeatedField<MxStatusProxy> Statuses { get; }
     }
 }

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

@@ -73,6 +73,19 @@ public sealed class MxAccessWriteCompletionCache
             ulong version = entries.TryGetValue(key, out CompletionEntry existing)
                 ? existing.Version + 1
                 : 1UL;
+
+            // Still a defensive copy, deliberately — this is NOT an oversight
+            // left behind by the borrow that MxAccessValueCache.Set adopted (see
+            // the ownership contract there). The value cache is handed the whole
+            // enqueued MxEvent, so the queue's write-once ownership invariant
+            // covers everything it retains. This method is handed a bare
+            // RepeatedField whose provenance its signature cannot constrain:
+            // the production sink does pass an enqueued event's Statuses, but
+            // callers that build and keep their own rows are equally valid
+            // against this API, and a borrowed alias would then let a later
+            // mutation rewrite an already-recorded completion. The write path is
+            // command-rate, not the per-OnDataChange streaming hot path, so the
+            // clone costs nothing worth reclaiming.
             entries[key] = new CompletionEntry(version, statuses.Clone());
         }