2026-07-09 09:58:57 -04:00
42 changed files with 1106 additions and 91 deletions
@@ -76,7 +76,7 @@ powershell -ExecutionPolicy Bypass -File scripts/run-client-e2e-tests.ps1
 - **Style guides** in `docs/style-guides/` are authoritative. Follow `CSharpStyleGuide.md` for gateway/worker/.NET-client code: file-scoped namespaces, `sealed` by default, `Async` suffix on Task-returning methods, MXAccess-aligned names (`MxStatusProxy`, `ServerHandle`, `ItemHandle`, `HResult`).
 - **MXAccess parity is the contract.** Don't "fix" surprising MXAccess behavior (e.g., `WriteSecured` failing before a value-bearing NMX body, distinct `OperationComplete` semantics, invalid-handle exceptions) unless the client explicitly opts into a non-parity mode. The installed MXAccess COM component is the baseline.
 - **Don't synthesize events.** The gateway forwards only events the worker emits; it never invents `OperationComplete` from write completion or command replies.
- **One worker per session** (invariant). Multi-subscriber event fan-out and reconnect-with-replay have shipped and are config-gated: `AllowMultipleEventSubscribers` (default `false`) enables fan-out up to `MaxEventSubscribersPerSession` (default `8`); `DetachGraceSeconds` (default `30`) retains a session after its last subscriber drops so clients can reconnect; `ReplayBufferCapacity` / `ReplayRetentionSeconds` control how much event history the replay ring keeps. Default config preserves the original single-subscriber, no-retention behavior. See `docs/DesignDecisions.md` and `docs/Sessions.md`.
+- **One worker per session** (invariant). Multi-subscriber event fan-out and reconnect-with-replay have shipped and are config-gated: `AllowMultipleEventSubscribers` (default `false`) enables fan-out up to `MaxEventSubscribersPerSession` (default `8`); `DetachGraceSeconds` (default `30`) retains a session after its last subscriber drops so clients can reconnect; `ReplayBufferCapacity` / `ReplayRetentionSeconds` control how much event history the replay ring keeps. Default config is single-subscriber (`AllowMultipleEventSubscribers` off), but detach-grace and replay retention are **on** by default (`DetachGraceSeconds=30`, `ReplayBufferCapacity=1024`, `ReplayRetentionSeconds=300`): a detached session is retained for 30 s and recent events are buffered for reconnect. See `docs/DesignDecisions.md` and `docs/Sessions.md`.
 - **Gateway restart does not reattach orphan workers.** The first version terminates orphaned workers on startup; do not design code paths that assume reattachment.
 - **No Blazor UI component libraries.** Dashboard uses local Bootstrap CSS/JS only — do not introduce MudBlazor, Radzen, FluentUI, etc.
 - **Don't log secrets or full tag values by default.** API keys, passwords, `WriteSecured` payloads, and `AuthenticateUser` credentials must never reach logs. Value logging is opt-in and redacted.
@@ -121,6 +121,8 @@ External analysis sources referenced by design docs:
 Gateway gRPC clients authenticate with an API key in metadata: `authorization: Bearer mxgw_<key-id>_<secret>`. Keys are stored hashed (with a peppered SHA) in a gateway-owned SQLite DB (default `C:\ProgramData\MxGateway\gateway-auth.db`). Scopes (`session`, `invoke`, `event`, `metadata`, `admin`) gate specific RPCs; missing → `Unauthenticated`, insufficient → `PermissionDenied`. The `apikey` subcommand on the server exe manages keys; see `src/ZB.MOM.WW.MxGateway.Server/Security/Authentication/`.
 Session event streaming is **owner-scoped**: the API key that opened a session is recorded on the session, and every `StreamEvents` attach/reattach is rejected with `PermissionDenied` unless the caller's key id matches the owner. Possessing the `event` scope and knowing a session id is not sufficient — this closes the reconnect/fan-out trust boundary (detach-grace and replay retention are on by default) so an `event`-scoped key cannot attach to another key's retained session.
 Dashboard auth is LDAP-backed (separate from the gRPC API-key model). `/login` binds against `MxGateway:Ldap` and maps the user's LDAP groups to `Admin` or `Viewer` via `MxGateway:Dashboard:GroupToRole`, then issues an HTTP-only secure `__Host-MxGatewayDashboard` cookie. SignalR hubs at `/hubs/{snapshot,alarms,events}` accept either the cookie or a 30-minute bearer minted at `/hubs/token`. `Dashboard:AllowAnonymousLocalhost` bypasses auth on loopback when enabled. `Dashboard:DisableLogin` (default `false`) auto-authenticates every dashboard request — including remote browsers — as `Dashboard:AutoLoginUser` (default `multi-role`) with both Admin and Viewer roles; dev/test only, never enable in production.
 ## Process / Platform Notes
@@ -42,14 +42,14 @@ Tiers follow the roadmap in `00-overall.md`. Work top-down; within a tier, `Dep`
 | ID | Sev | Eff | Dep | Status | Title |
 |---|---|:-:|---|---|---|
-| GWC-01 | Critical | M | — | Not started | Alarm monitor and distributor pump both drain the single worker event channel |
+| GWC-01 | Critical | M | — | Done | Alarm monitor and distributor pump both drain the single worker event channel |
-| GWC-02 | High | M | — | Not started | Faulted sessions are never swept (worker + slot pinned up to 30 min) |
+| GWC-02 | High | M | — | Done | Faulted sessions are never swept (worker + slot pinned up to 30 min) |
-| GWC-03 | High | S | — | Not started | Documented sparse-array max-length bound is unimplemented |
+| GWC-03 | High | S | — | Done | Documented sparse-array max-length bound is unimplemented |
-| WRK-01 | High | M | — | Not started | Long `ReadBulk` self-faults as `StaHung`; all replies then dropped |
+| WRK-01 | High | M | — | Done | Long `ReadBulk` self-faults as `StaHung`; all replies then dropped |
-| CLI-01 | High | M | — | Not started | Go `Session.Events()` silently closes stream on 16-slot overflow |
+| CLI-01 | High | M | — | Done | Go `Session.Events()` silently closes stream on 16-slot overflow |
-| CLI-03 | High | M | — | Not started | Rust `invoke` never validates HRESULT / MXSTATUS_PROXY |
+| CLI-03 | High | M | — | Done | Rust `invoke` never validates HRESULT / MXSTATUS_PROXY |
-| TST-02 | High | M | TST-04 | Not started | Reconnect owner re-validation not implemented (security control) |
+| TST-02 | High | M | TST-04 | Done | Reconnect owner re-validation not implemented (security control) |
-| TST-12 | Medium | S | — | Not started | CLAUDE.md misstates default retention behaviour (defaults are on, not off) |
+| TST-12 | Medium | S | — | Done | CLAUDE.md misstates default retention behaviour (defaults are on, not off) |
 ## Finding registers by domain
@@ -59,9 +59,9 @@ Full design + implementation for each row lives in the linked domain doc under i
 | ID | Sev | Tier | Eff | Dep | Status | Title |
 |---|---|:-:|:-:|---|---|---|
-| GWC-01 | Critical | P0 | M | — | Not started | Alarm monitor and distributor pump both drain the single worker event channel |
+| GWC-01 | Critical | P0 | M | — | Done | Alarm monitor and distributor pump both drain the single worker event channel |
-| GWC-02 | High | P0 | M | — | Not started | Faulted sessions are never swept |
+| GWC-02 | High | P0 | M | — | Done | Faulted sessions are never swept |
-| GWC-03 | High | P0 | S | — | Not started | Documented sparse-array max-length bound is unimplemented |
+| GWC-03 | High | P0 | S | — | Done | Documented sparse-array max-length bound is unimplemented |
 | GWC-04 | Medium | P1 | M | — | Not started | Full event channel stalls the worker read loop behind command replies |
 | GWC-05 | Medium | — | S | — | Not started | Worker pipe created with no ACL / no CurrentUserOnly |
 | GWC-06 | Medium | P2 | S | GWC-07,08 | Not started | Stopwatch allocated per streamed event |
@@ -87,7 +87,7 @@ Full design + implementation for each row lives in the linked domain doc under i
 | ID | Sev | Tier | Eff | Dep | Status | Title |
 |---|---|:-:|:-:|---|---|---|
-| WRK-01 | High | P0 | M | — | Not started | Long `ReadBulk` self-faults as `StaHung`; all replies then dropped |
+| WRK-01 | High | P0 | M | — | Done | Long `ReadBulk` self-faults as `StaHung`; all replies then dropped |
 | WRK-02 | Medium | — | M | — | Not started | STA thread death after startup is silent; future work hangs forever |
 | WRK-03 | Medium | — | S | WRK-02 | Not started | Commands after shutdown starts are dropped with no reply |
 | WRK-04 | Medium | — | M | — | Not started | Envelope `sequence` can appear out of order on the wire |
@@ -174,9 +174,9 @@ Full design + implementation for each row lives in the linked domain doc under i
 | ID | Sev | Tier | Eff | Dep | Status | Title |
 |---|---|:-:|:-:|---|---|---|
-| CLI-01 | High | P0 | M | — | Not started | Go `Session.Events()` silently closes stream on 16-slot overflow |
+| CLI-01 | High | P0 | M | — | Done | Go `Session.Events()` silently closes stream on 16-slot overflow |
 | CLI-02 | High | P1 | M | — | Not started | Rust crate unbuildable outside the repo (`build.rs` path + `--no-verify`) |
-| CLI-03 | High | P0 | M | — | Not started | Rust `invoke` never validates HRESULT / MXSTATUS_PROXY |
+| CLI-03 | High | P0 | M | — | Done | Rust `invoke` never validates HRESULT / MXSTATUS_PROXY |
 | CLI-04 | High | P2 | L | CLI-15 | Not started | Typed-command parity gap across all five clients |
 | CLI-05 | Medium | — | S | — | Not started | .NET session cannot be re-attached to an existing session id |
 | CLI-06 | Medium | — | S | — | Not started | .NET `DisposeAsync` blocks/throws on unreachable gateway |
@@ -214,7 +214,7 @@ Full design + implementation for each row lives in the linked domain doc under i
 | ID | Sev | Tier | Eff | Dep | Status | Title |
 |---|---|:-:|:-:|---|---|---|
 | TST-01 | High | P2 | L | TST-04 | Not started | Reconnect/replay has no e2e test and no client consumer |
-| TST-02 | High | P0 | M | TST-04 | Not started | Reconnect owner re-validation not implemented |
+| TST-02 | High | P0 | M | TST-04 | Done | Reconnect owner re-validation not implemented |
 | TST-03 | High | P1 | M | — | Not started | No CI exists |
 | TST-04 | High | P2 | L | — | Not started | Session-resilience epic 16/28 tasks unfinished |
 | TST-05 | Medium | P1 | S | TST-03 | Not started | Real-worker control/COM paths verified opt-in only |
@@ -224,7 +224,7 @@ Full design + implementation for each row lives in the linked domain doc under i
 | TST-09 | Medium | — | M | — | Not started | Health checks cover only the auth store |
 | TST-10 | Medium | — | M | — | Not started | Deployment/upgrade undocumented and hand-rolled |
 | TST-11 | Medium | P2 | M | — | Not started | No version discipline on server; client versions drift |
-| TST-12 | Medium | P0 | S | — | Not started | CLAUDE.md misstates default retention behaviour |
+| TST-12 | Medium | P0 | S | — | Done | CLAUDE.md misstates default retention behaviour |
 | TST-13 | Medium | P2 | S | — | Not started | gateway.md carries stale design-era sketches |
 | TST-14 | Medium | P2 | S | — | Not started | Repo-root working artifacts need triage |
 | TST-15 | Medium | P2 | M | TST-04 | Not started | Dashboard EventsHub has no per-session ACL |
@@ -254,3 +254,5 @@ Findings the review flagged as one coordinated design pass — sequence them tog
 | Date | Change |
 |---|---|
 | 2026-07-09 | Initial tracking doc generated from the six domain remediation designs. All 153 findings `Not started`. |
 | 2026-07-09 | P0 tier executed via parallel agents. GWC-01/02/03, TST-02, TST-12, CLI-01, CLI-03 → `Done` (NonWindows build clean; gateway 135/135 targeted tests pass, Go `go test ./...` + `-race` pass, Rust `test`/`clippy -D warnings` pass). |
 | 2026-07-09 | WRK-01 → `Done`. Verified on Windows host (windev) via an isolated `origin/main` worktree: worker builds x86 clean, `StaRuntimeTests`+`WorkerPipeSessionTests` 33/33 pass. Fixed an `xUnit1030` build error (the new worker test used `.ConfigureAwait(false)` in `[Fact]` bodies) that the macOS tree could not surface. Also ran GWC-01's Windows-only `WorkerClientTests` on windev: 18/18 pass (incl. `ReadEventsAsync_SecondEnumerator_Throws`). All 8 P0 findings now `Done`. Not yet committed. |
@@ -147,6 +147,59 @@ func TestEventsAfterCancelsStreamWhenCompatibilityChannelIsAbandoned(t *testing.
 	}
 }
 func TestEventsSlowConsumerYieldsErrSlowConsumerBeforeClose(t *testing.T) {
 	fake := &fakeGatewayServer{
 		streamStarted:    make(chan struct{}),
 		streamDone:       make(chan struct{}),
 		streamEventCount: 64,
 	}
 	client, cleanup := newBufconnClient(t, fake)
 	defer cleanup()
 	session := NewSessionForID(client, "session-1")
 	events, err := session.EventsAfter(context.Background(), 0)
 	if err != nil {
 		t.Fatalf("EventsAfter() error = %v", err)
 	}
 	<-fake.streamStarted
 	// Do not drain the channel so the buffer overflows. The stream stops once
 	// the slow-consumer cancel fires on the producer side.
 	select {
 	case <-fake.streamDone:
 	case <-time.After(2 * time.Second):
 		t.Fatal("event stream did not stop after buffer overflow")
 	}
 	var last EventResult
 	gotResult := false
 	for {
 		select {
 		case res, ok := <-events:
 			if !ok {
 				if !gotResult {
 					t.Fatal("events channel closed without yielding any result")
 				}
 				if !errors.Is(last.Err, ErrSlowConsumer) {
 					t.Fatalf("final event result err = %v, want one wrapping ErrSlowConsumer", last.Err)
 				}
 				var gwErr *GatewayError
 				if !errors.As(last.Err, &gwErr) {
 					t.Fatalf("final event result err is %T, want *GatewayError", last.Err)
 				}
 				if gwErr.Op != "stream events" {
 					t.Fatalf("final event result err Op = %q, want %q", gwErr.Op, "stream events")
 				}
 				return
 			}
 			last = res
 			gotResult = true
 		case <-time.After(2 * time.Second):
 			t.Fatal("events channel did not close after slow-consumer termination")
 		}
 	}
 }
 func TestSessionHelpersBuildCommandsAndExposeRawReply(t *testing.T) {
 	fake := &fakeGatewayServer{
 		invokeReply: &pb.MxCommandReply{
@@ -1,11 +1,19 @@
 package mxgateway
 import (
 	"errors"
 	"fmt"
 	pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
 )
 // ErrSlowConsumer is the terminal error sent on the Events/EventsAfter
 // (cancel-when-full) path when the buffered results channel overflows because
 // the consumer fell behind. It is delivered as the final EventResult.Err before
 // the channel closes, so overflow is always observable rather than silently
 // dropping events. Match it with errors.Is.
 var ErrSlowConsumer = errors.New("mxgateway: event consumer fell behind; stream terminated")
 // GatewayError wraps transport-level gRPC failures.
 type GatewayError struct {
 	// Op names the operation that failed (for example "dial" or "invoke").
@@ -18,6 +18,15 @@ import (
 const maxBulkItems = 1000
 // eventBufferSize is the number of buffered event slots on the
 // Events/EventsAfter (cancel-when-full) results channel.
 const eventBufferSize = 16
 // eventBufferReservedSlots is the extra capacity reserved beyond
 // eventBufferSize so a terminal ErrSlowConsumer result can always be enqueued
 // non-blockingly on overflow, even when all data slots are full.
 const eventBufferReservedSlots = 1
 // EventResult carries either the next ordered event or a terminal stream error.
 type EventResult struct {
 	// Event is the next event from the stream when Err is nil.
@@ -674,11 +683,22 @@ func (s *Session) Write2Raw(ctx context.Context, serverHandle, itemHandle int32,
 // Events streams ordered session events until the server ends the stream,
 // context cancellation stops Recv, or a terminal error is sent.
 //
 // The returned channel is buffered. If the consumer falls behind and the buffer
 // overflows, the stream is terminated and a final EventResult carrying a
 // GatewayError that wraps ErrSlowConsumer is delivered before the channel
 // closes. Callers must match it with errors.Is(res.Err, ErrSlowConsumer) to
 // distinguish a slow-consumer drop from a graceful server end. Use
 // SubscribeEvents for a blocking, backpressured stream that never drops.
 func (s *Session) Events(ctx context.Context) (<-chan EventResult, error) {
 	return s.EventsAfter(ctx, 0)
 }
 // EventsAfter streams ordered session events after the given worker sequence.
 //
 // Like Events, the returned channel is buffered and terminates with a final
 // EventResult wrapping ErrSlowConsumer (matchable via errors.Is) if the consumer
 // falls behind and the buffer overflows, rather than silently closing.
 func (s *Session) EventsAfter(ctx context.Context, afterWorkerSequence uint64) (<-chan EventResult, error) {
 	subscription, err := s.subscribeEventsAfter(ctx, afterWorkerSequence, true)
 	if err != nil {
@@ -708,7 +728,7 @@ func (s *Session) subscribeEventsAfter(ctx context.Context, afterWorkerSequence
 		return nil, err
 	}
-	results := make(chan EventResult, 16)
+	results := make(chan EventResult, eventBufferSize+eventBufferReservedSlots)
 	done := make(chan struct{})
 	go func() {
 		defer close(results)
@@ -756,14 +776,30 @@ func sendEventResult(
 	cancel context.CancelFunc,
 ) bool {
 	if cancelWhenBufferFull {
 		// Treat the channel as full once the eventBufferSize data slots are
 		// occupied, keeping eventBufferReservedSlots free for the terminal
 		// error. This goroutine is the sole producer, so len(results) only
 		// grows by our own sends and shrinks as the consumer reads; the reserve
 		// therefore always survives to carry ErrSlowConsumer. A plain buffered
 		// send would instead consume the reserved slot as ordinary data.
 		if len(results) >= eventBufferSize {
 			// The consumer fell behind and the data slots are full. Cancel the
 			// stream, then use the reserved terminal slot to enqueue a loud
 			// ErrSlowConsumer result so overflow is always observable rather
 			// than a silently closed channel indistinguishable from a graceful
 			// server end.
 			cancel()
 			select {
 			case results <- EventResult{Err: &GatewayError{Op: "stream events", Err: ErrSlowConsumer}}:
 			default:
 			}
 			return false
 		}
 		select {
 		case results <- result:
 			return true
 		case <-ctx.Done():
 			return false
 		default:
 			cancel()
 			return false
 		}
 	}
@@ -125,6 +125,18 @@ preserving the raw message for parity diagnostics. Command replies whose
 protocol status is not `PROTOCOL_STATUS_CODE_OK` become `Error::Command` and
 retain the raw `MxCommandReply`.
 The typed command helpers (`register`, `add_item`, `write`, the bulk variants,
 etc.) also enforce MXAccess parity on an otherwise-OK reply: a reply that
 reports a negative `hresult` (COM failure semantics — a positive code such as
 `S_FALSE = 1` is a success) or a non-success `MXSTATUS_PROXY` status entry
 becomes `Error::MxAccess`, which boxes an `MxAccessError` retaining the raw
 `MxCommandReply` (recover it with `MxAccessError::reply` / `into_reply`). Its
 message summarizes the `hresult` and status entries with credential-safe
 redaction. Per-item bulk failures are reported inside each result entry
 (`was_successful = false`) and do not raise `Error::MxAccess`. The raw
 `invoke_raw` / `client.invoke_raw` escape hatch performs neither check and
 returns the unvalidated reply.
 ## Write Semantics And Common Pitfalls
 These are MXAccess parity behaviors that surprise new callers. The gateway
@@ -11,7 +11,9 @@ use tonic::transport::Channel;
 use tonic::Request;
 use crate::auth::AuthInterceptor;
-use crate::error::{ensure_command_success, ensure_protocol_success, Error};
+use crate::error::{
    ensure_command_success, ensure_mxaccess_success, ensure_protocol_success, Error,
 };
 use crate::generated::mxaccess_gateway::v1::mx_access_gateway_client::MxAccessGatewayClient;
 use crate::generated::mxaccess_gateway::v1::{
    AcknowledgeAlarmReply, AcknowledgeAlarmRequest, ActiveAlarmSnapshot, AlarmFeedMessage,
@@ -166,16 +168,24 @@ impl GatewayClient {
        Ok(response.into_inner())
    }
-    /// Issue an `Invoke` RPC and surface a non-OK reply as
+    /// Issue an `Invoke` RPC and surface a failing reply as a typed error.
-    /// [`Error::Command`].
+    ///
    /// The reply is validated twice: [`ensure_command_success`] rejects a
    /// non-OK protocol envelope as [`Error::Command`], then
    /// [`ensure_mxaccess_success`] rejects an MXAccess-level failure (negative
    /// `hresult` or a non-success `MXSTATUS_PROXY` entry) as
    /// [`Error::MxAccess`], preserving MXAccess parity. Callers that need the
    /// unvalidated reply should use [`invoke_raw`](Self::invoke_raw) instead.
    ///
    /// # Errors
    ///
    /// Returns [`Error::Command`] when the reply's `protocol_status` is not
-    /// `Ok`, plus any errors propagated by
+    /// `Ok`, [`Error::MxAccess`] when the reply reports an MXAccess-level
    /// failure, plus any errors propagated by
    /// [`invoke_raw`](Self::invoke_raw).
    pub async fn invoke(&self, request: MxCommandRequest) -> Result<MxCommandReply, Error> {
-        ensure_command_success(self.invoke_raw(request).await?)
+        let reply = ensure_command_success(self.invoke_raw(request).await?)?;
        ensure_mxaccess_success(reply)
    }
    /// Open the server-streaming `StreamEvents` RPC.
@@ -9,7 +9,9 @@
 use thiserror::Error as ThisError;
 use tonic::Code;
-use crate::generated::mxaccess_gateway::v1::{MxCommandReply, ProtocolStatus, ProtocolStatusCode};
+use crate::generated::mxaccess_gateway::v1::{
    MxCommandReply, MxStatusCategory, ProtocolStatus, ProtocolStatusCode,
 };
 /// Top-level error type returned by the Rust client wrappers.
 ///
@@ -95,6 +97,15 @@ pub enum Error {
    #[error("gateway command failed: {0}")]
    Command(#[from] Box<CommandError>),
    /// Gateway accepted the call and returned an `Ok` protocol envelope, but
    /// the reply reported an MXAccess-level failure — a negative `hresult`
    /// (COM failure semantics) or one or more `MXSTATUS_PROXY` entries that
    /// did not indicate success. The wrapped [`MxAccessError`] preserves the
    /// full reply so callers can inspect the native status payload. Boxed to
    /// keep the containing enum small, matching [`Error::Command`].
    #[error("gateway command reported an MXAccess failure: {0}")]
    MxAccess(#[from] Box<MxAccessError>),
    /// Protocol-level operation (open/close session) returned a non-OK
    /// [`ProtocolStatus`] envelope.
    #[error("gateway {operation} failed: {code:?}: {message}")]
@@ -175,6 +186,72 @@ impl std::fmt::Display for CommandError {
 impl std::error::Error for CommandError {}
 /// Wrapper around an [`MxCommandReply`] whose protocol envelope succeeded but
 /// whose MXAccess-level result reported a failure — a negative `hresult` or a
 /// non-success `MXSTATUS_PROXY` entry.
 ///
 /// The wrapper is heap-allocated inside [`Error::MxAccess`] to keep the
 /// containing enum small. Callers can recover the reply with
 /// [`MxAccessError::reply`] or [`MxAccessError::into_reply`]. Its `Display`
 /// summarizes the `hresult` and status entries and scrubs any credential-like
 /// tokens from diagnostic text before it reaches a caller.
 #[derive(Clone, Debug)]
 pub struct MxAccessError {
    reply: MxCommandReply,
 }
 impl MxAccessError {
    /// Wrap a reply whose MXAccess-level result reported a failure.
    pub fn new(reply: MxCommandReply) -> Self {
        Self { reply }
    }
    /// Borrow the underlying reply (correlation id, hresult, statuses).
    pub fn reply(&self) -> &MxCommandReply {
        &self.reply
    }
    /// Consume the error and return the underlying reply.
    pub fn into_reply(self) -> MxCommandReply {
        self.reply
    }
 }
 impl std::fmt::Display for MxAccessError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let hresult = match self.reply.hresult {
            Some(value) => value.to_string(),
            None => "none".to_owned(),
        };
        write!(
            formatter,
            "hresult={hresult}, {} status entr{}",
            self.reply.statuses.len(),
            if self.reply.statuses.len() == 1 {
                "y"
            } else {
                "ies"
            }
        )?;
        for status in &self.reply.statuses {
            let category = MxStatusCategory::try_from(status.category)
                .unwrap_or(MxStatusCategory::Unspecified);
            let diagnostic = redact_credentials(&status.diagnostic_text);
            write!(
                formatter,
                "; [success={}, category={category:?}, detail={}, {}]",
                status.success, status.detail, diagnostic
            )?;
        }
        Ok(())
    }
 }
 impl std::error::Error for MxAccessError {}
 impl From<tonic::Status> for Error {
    fn from(status: tonic::Status) -> Self {
        let message = redact_credentials(status.message());
@@ -225,6 +302,36 @@ pub fn ensure_command_success(reply: MxCommandReply) -> Result<MxCommandReply, E
    }
 }
 /// Promote an MXAccess-level failure carried inside an otherwise-successful
 /// [`MxCommandReply`] to an [`Error::MxAccess`].
 ///
 /// This is the second reply check applied to the typed command path, after
 /// [`ensure_command_success`] confirms the protocol envelope is `Ok`. It
 /// enforces MXAccess parity: a reply can carry an `Ok` protocol envelope while
 /// MXAccess itself rejected the operation. Following COM semantics (and the
 /// Python client), only a **negative** `hresult` is a failure — positive codes
 /// such as `S_FALSE = 1` are success. A `MXSTATUS_PROXY` entry is treated as a
 /// failure when its `success` member is `0`.
 ///
 /// Per-item bulk failures are reported inside each result entry
 /// (`was_successful = false`) rather than in the top-level `hresult`/`statuses`
 /// fields, so this check does not trip on them — matching the other clients.
 ///
 /// # Errors
 ///
 /// Returns [`Error::MxAccess`] when `reply.hresult` is negative or any
 /// `reply.statuses` entry reports a non-success `success` member.
 pub fn ensure_mxaccess_success(reply: MxCommandReply) -> Result<MxCommandReply, Error> {
    let hresult_failure = reply.hresult.is_some_and(|hresult| hresult < 0);
    let status_failure = reply.statuses.iter().any(|status| status.success == 0);
    if hresult_failure || status_failure {
        Err(Box::new(MxAccessError::new(reply)).into())
    } else {
        Ok(reply)
    }
 }
 /// Validate a [`ProtocolStatus`] envelope returned by an open/close-session
 /// reply.
 ///
@@ -271,7 +378,20 @@ fn redact_credentials(message: &str) -> String {
 mod tests {
    use tonic::{Code, Status};
-    use super::Error;
+    use super::{ensure_mxaccess_success, Error};
    use crate::generated::mxaccess_gateway::v1::{
        MxCommandReply, MxStatusCategory, MxStatusProxy, ProtocolStatus, ProtocolStatusCode,
    };
    fn ok_reply() -> MxCommandReply {
        MxCommandReply {
            protocol_status: Some(ProtocolStatus {
                code: ProtocolStatusCode::Ok as i32,
                message: String::new(),
            }),
            ..MxCommandReply::default()
        }
    }
    #[test]
    fn classifies_authentication_status() {
@@ -286,4 +406,48 @@ mod tests {
        assert!(message.contains("<redacted>"));
        assert!(!message.contains("visible_secret"));
    }
    #[test]
    fn ensure_mxaccess_success_passes_clean_reply() {
        let mut reply = ok_reply();
        // Positive hresult (e.g. S_FALSE = 1) is a success, not a failure.
        reply.hresult = Some(1);
        reply.statuses = vec![MxStatusProxy {
            success: 1,
            category: MxStatusCategory::Ok as i32,
            ..MxStatusProxy::default()
        }];
        assert!(ensure_mxaccess_success(reply).is_ok());
    }
    #[test]
    fn ensure_mxaccess_success_flags_failing_status_entry() {
        let mut reply = ok_reply();
        reply.statuses = vec![MxStatusProxy {
            success: 0,
            category: MxStatusCategory::CommunicationError as i32,
            detail: 42,
            diagnostic_text: "write rejected for mxgw_visible_secret".to_owned(),
            ..MxStatusProxy::default()
        }];
        let error = ensure_mxaccess_success(reply).expect_err("failing status must error");
        let message = error.to_string();
        assert!(matches!(error, Error::MxAccess(_)));
        assert!(message.contains("<redacted>"));
        assert!(!message.contains("visible_secret"));
    }
    #[test]
    fn ensure_mxaccess_success_flags_negative_hresult() {
        let mut reply = ok_reply();
        // 0x80004005 (E_FAIL) as a signed 32-bit value.
        reply.hresult = Some(-2_147_467_259);
        let error = ensure_mxaccess_success(reply).expect_err("negative hresult must error");
        assert!(matches!(error, Error::MxAccess(_)));
    }
 }
@@ -26,7 +26,7 @@ pub use auth::{ApiKey, AuthInterceptor};
 #[doc(inline)]
 pub use client::{AlarmFeedStream, EventStream, GatewayClient};
 #[doc(inline)]
-pub use error::{CommandError, Error};
+pub use error::{CommandError, Error, MxAccessError};
 #[doc(inline)]
 pub use galaxy::{DeployEventStream, GalaxyClient};
 #[doc(inline)]
@@ -85,6 +85,17 @@ delivery. If the requested position precedes the oldest retained event, a
 is atomic (no gap, no duplicate). See [Sessions](./Sessions.md) for the full
 reconnect and replay protocol.
 Decision: event-stream attach is bound to the opening API key. Because the
 detach-grace and replay-retention windows are on by default, the reconnect
 surface is a trust boundary — a retained session outlives the stream that opened
 it. The session records the opening key id (`GatewaySession.OwnerKeyId`) and
 every `StreamEvents` attach/reattach is rejected with `PermissionDenied` unless
 the caller's key id matches the owner. Gating on the `event` scope alone was
 rejected: it would let any `event`-scoped key that learned a session id attach to
 another key's retained session and receive its replayed and live data. The check
 runs before any subscriber is attached, so a foreign key never touches the
 replay ring. Admin-scope override is deferred.
 ## Event Subscribers
 Multi-subscriber fan-out for data-side `StreamEvents` is shipped and
@@ -38,6 +38,7 @@ paths, timeouts, queue sizes, enum values, or protocol values are invalid.
      "DefaultLeaseSeconds": 1800,
      "LeaseSweepIntervalSeconds": 30,
      "DetachGraceSeconds": 30,
      "FaultedGraceSeconds": 0,
      "AllowMultipleEventSubscribers": false,
      "MaxEventSubscribersPerSession": 8,
      "WorkerReadyWaitTimeoutMs": 0
@@ -46,7 +47,8 @@ paths, timeouts, queue sizes, enum values, or protocol values are invalid.
      "QueueCapacity": 10000,
      "BackpressurePolicy": "FailFast",
      "ReplayBufferCapacity": 1024,
-      "ReplayRetentionSeconds": 300
+      "ReplayRetentionSeconds": 300,
      "MaxSparseArrayLength": 1000000
    },
    "Dashboard": {
      "Enabled": true,
@@ -129,6 +131,7 @@ to avoid accidental large allocations from malformed or oversized frames.
 | `MxGateway:Sessions:DefaultLeaseSeconds` | `1800` | Initial session lease and refresh duration. Unary client activity extends the lease by this duration. |
 | `MxGateway:Sessions:LeaseSweepIntervalSeconds` | `30` | Hosted monitor interval for closing expired leases. Active event-stream subscribers keep a session from expiring while the stream remains attached. |
 | `MxGateway:Sessions:DetachGraceSeconds` | `30` | Detach-grace retention window. When positive, a session whose last external (gRPC) event-stream subscriber drops is retained in `Ready` for this many seconds so a client can reconnect; if no external subscriber re-attaches within the window, the lease monitor closes it with `detach-grace-expired`. The internal dashboard mirror does not count as an external subscriber, so a dashboard-only session still enters detach-grace. `0` disables retention and reverts to closing only on normal lease expiry. Must be zero or greater. Reconnect/replay itself is implemented separately (Task 12); this option controls retention and expiry only. The effective close happens within the next sweep cycle after the window elapses — up to `LeaseSweepIntervalSeconds` after expiry. Operators wanting a firm minimum retention bound should set `DetachGraceSeconds` greater than `LeaseSweepIntervalSeconds`. |
 | `MxGateway:Sessions:FaultedGraceSeconds` | `0` | Grace window before the lease monitor reaps a faulted session (killing its worker and freeing the slot). A faulted session is permanently unusable yet otherwise pins a session slot and a live x86 worker until its normal lease expires. `0` (the default) reaps it on the next sweep cycle, bounding blast radius; a positive value keeps the faulted session observable via `GetSessionStatus` for that window before it is reclaimed, with the close reason `faulted-reaped`. Must be zero or greater. |
 | `MxGateway:Sessions:AllowMultipleEventSubscribers` | `false` | Controls whether multiple `StreamEvents` subscribers may attach to one session. When `false` the session refuses a second subscriber with `AlreadyExists`. Set to `true` to enable fan-out via the `SessionEventDistributor`. |
 | `MxGateway:Sessions:MaxEventSubscribersPerSession` | `8` | Maximum number of concurrent `StreamEvents` subscribers per session when `AllowMultipleEventSubscribers` is `true`. Effectively 1 when `AllowMultipleEventSubscribers` is `false`. Must be greater than zero. |
 | `MxGateway:Sessions:WorkerReadyWaitTimeoutMs` | `0` | Bounded time, in milliseconds, the gateway will wait for a worker to reach `Ready` when the session is already `Ready` but the worker state has transiently diverged (e.g. `Handshaking` after a heartbeat blip). Applies only to transient worker states; terminal states (`Faulted`/`Closing`/`Closed`/no worker) fail fast immediately regardless of this setting. `0` (the default) disables the wait and preserves the original fail-fast behavior. Must be greater than or equal to zero. |
@@ -143,6 +146,7 @@ All numeric session options must be greater than zero.
 | `MxGateway:Events:BackpressurePolicy` | `FailFast` | Per-subscriber event backpressure behavior when a subscriber's bounded event channel overflows. Overflow is isolated to the offending subscriber: it is always disconnected with an `EventQueueOverflow` fault while the session pump and other subscribers keep running. `FailFast` additionally faults the whole session only in the legacy single-subscriber case (the current default mode); with multiple subscribers it degrades to a per-subscriber disconnect so one slow consumer never faults a shared session. `DisconnectSubscriber` disconnects only the slow subscriber in all cases. |
 | `MxGateway:Events:ReplayBufferCapacity` | `1024` | Maximum number of events retained per session in the replay ring buffer, used to re-deliver events a returning subscriber missed (reconnect/reattach). The oldest retained event is evicted once this count is exceeded. `0` disables replay retention. |
 | `MxGateway:Events:ReplayRetentionSeconds` | `300` | Maximum age, in seconds, of an event retained in the replay ring buffer. Entries older than this are evicted regardless of capacity. `0` disables age-based eviction. |
 | `MxGateway:Events:MaxSparseArrayLength` | `1000000` | Maximum `total_length` a sparse-array write (`MxSparseArray`) may declare. A write above this cap is rejected with `InvalidArgument` before the full array is materialized, guarding against a single write forcing a multi-GB allocation. Must be between `1` and `Array.MaxLength`. |
 `QueueCapacity` must be greater than zero; it bounds each per-subscriber event
 channel fed by the session's single event pump. A slow subscriber overflows only
@@ -656,11 +656,15 @@ the event queue implementation owns those counters.
 The STA watchdog currently emits a `WorkerFault` with
 `WorkerFaultCategory.StaHung` when `LastStaActivityUtc` is older than
 `WorkerPipeSessionOptions.HeartbeatGrace` **and no command is in flight**.
-`StaRuntime.ProcessQueuedCommands` calls `MarkActivity()` only immediately
+`StaRuntime.ProcessQueuedCommands` calls `MarkActivity()` immediately before
-before and after each work item, so a synchronously long-running STA command
+and after each work item, so a synchronously long-running STA command that
-(for example a `ReadBulk` waiting `timeout_ms` for the first `OnDataChange`)
+neither completes work items nor pumps would freeze `LastStaActivityUtc` for
-legitimately freezes `LastStaActivityUtc` for the duration of the wait while
+the duration of the wait while the worker is healthy. Commands that hold the
-the worker is healthy. The watchdog is therefore suppressed while the
+STA to wait for COM events (for example a `ReadBulk` waiting `timeout_ms` for
 the first `OnDataChange`) avoid this: they pump via
 `StaRuntime.PumpPendingMessages()`, which now refreshes `LastStaActivityUtc`
 on every iteration (see the `HeartbeatStuckCeiling` discussion below). The
 watchdog is additionally suppressed while the
 heartbeat snapshot's `CurrentCommandCorrelationId` is non-empty: the worker is
 busy executing a command, not hung, and the heartbeat already surfaces the
 in-flight correlation id so the gateway can apply its own per-command timeout
@@ -684,10 +688,18 @@ session and only the gateway's per-command timeout would catch the hang —
 losing the worker-originated diagnostic (`StaHung` fault category, the
 stale-by interval) from the gateway audit trail. Once `LastStaActivityUtc`
 has been stale for longer than `HeartbeatStuckCeiling`, the watchdog fires
-`StaHung` regardless of whether a command is in flight, on the assumption
+`StaHung` regardless of whether a command is in flight. This is now safe for
-that no legitimate STA command should run that long without periodically
+healthy long-running commands: `StaRuntime.PumpPendingMessages()` refreshes
-refreshing activity. Deployments that legitimately run very long bulk
+`LastStaActivityUtc` (via `MarkActivity()`) every time it runs, and long-hold
-operations should raise the ceiling rather than disable it.
+STA commands invoke it on every wait iteration (`ReadBulk` routes its
 per-tag wait through the `pumpStep` wired from `StaRuntime.PumpPendingMessages`).
 A command that keeps pumping therefore keeps its activity timestamp fresh and
 never reaches the ceiling, while a genuinely stuck STA — one that has stopped
 pumping — accrues staleness and faults correctly. The ceiling is thus the
 backstop for a command that both holds the thread and stops pumping, not a
 guillotine for slow-but-healthy work. Deployments that legitimately run very
 long bulk operations should still be able to raise the ceiling rather than
 disable it.
 ## Shutdown
@@ -72,7 +72,7 @@ private void EnsureSessionCapacity()
 }
 ```
-`SessionManager` also defines four close-reason constants — `DefaultCloseReason` (`"client-close"`), `GatewayShutdownReason` (`"gateway-shutdown"`), `LeaseExpiredReason` (`"lease-expired"`), and `DetachGraceExpiredReason` (`"detach-grace-expired"`) — so that the metrics and worker shutdown paths agree on a fixed vocabulary.
+`SessionManager` also defines five close-reason constants — `DefaultCloseReason` (`"client-close"`), `GatewayShutdownReason` (`"gateway-shutdown"`), `LeaseExpiredReason` (`"lease-expired"`), `FaultedReason` (`"faulted-reaped"`), and `DetachGraceExpiredReason` (`"detach-grace-expired"`) — so that the metrics and worker shutdown paths agree on a fixed vocabulary.
 ### SessionRegistry (ISessionRegistry)
@@ -197,6 +197,8 @@ Event streaming uses `AttachEventSubscriber` which returns a disposable lease. W
 `FailFast` event backpressure faults the whole session only in single-subscriber mode; in multi-subscriber mode it degrades to a per-subscriber disconnect so one slow consumer never faults a session shared by others. The session passes its mode to the `SessionEventDistributor` at construction, so this decision is made on the fixed mode rather than a live subscriber-count snapshot.
 The single worker event channel has exactly one direct reader: the `SessionEventDistributor` pump (`MapWorkerEventsAsync`). Both gateway-owned internal consumers — the dashboard mirror and the central alarm monitor — attach as distributor subscribers rather than draining the worker channel themselves. `GatewaySession.AttachInternalEventSubscriber` mirrors the dashboard-mirror lease (`isInternal: true`): the alarm monitor's `SessionManager.ReadAlarmEventsAsync` registers one so it consumes the same mapped `MxEvent`s the pump fans to every subscriber, without counting against `MaxEventSubscribersPerSession` and without a slow reconcile faulting the session. This is what keeps the alarm feed and the dashboard from splitting the stream between two raw drains (which would silently lose Acknowledge and provider-mode transitions); the worker channel is single-reader and a second `WorkerClient.ReadEventsAsync` consumer throws so a regression fails loudly.
 Sessions open with `MxGateway:Sessions:DefaultLeaseSeconds` (default 1800) added to the open timestamp. Unary client activity refreshes the lease by the same duration. `ExtendLease` and `IsLeaseExpired` cooperate with `SessionManager.CloseExpiredLeasesAsync`, which iterates a registry snapshot and closes any session whose lease has expired with `LeaseExpiredReason`. `SessionLeaseMonitorHostedService` runs that sweep every `MxGateway:Sessions:LeaseSweepIntervalSeconds` seconds (default 30).
 #### Detach-grace retention
@@ -207,10 +209,16 @@ Mechanically: when the last external subscriber detaches and `DetachGraceSeconds
 `DetachGraceSeconds` controls retention and expiry only; the reconnect/replay path that re-attaches a dropped client to a retained session is described in [Reconnect and replay](#reconnect-and-replay).
 #### Faulted-session reaping
 A session that faults (for example, a slow single-subscriber client that overflows its event queue under the `FailFast` backpressure policy) is left permanently unusable — every command fails `EvaluateReadyUnderLock` — but its worker keeps running and it still pins one of `MxGateway:Sessions:MaxSessions` slots. Rather than waiting up to `DefaultLeaseSeconds` for the lease to expire, `SessionManager.CloseExpiredLeasesAsync` also checks `IsFaultedReapable(now)` and reaps a faulted session through the same `TryBeginCloseIfExpired` → `CloseSessionCoreAsync` teardown, with the distinct `FaultedReason` (`"faulted-reaped"`). `MarkFaulted` stamps a fault timestamp from the session's `TimeProvider`; `MxGateway:Sessions:FaultedGraceSeconds` (default `0`) bounds how long the faulted session is retained before the next sweep reaps it — `0` reaps it on the next sweep, a positive value keeps it observable via `GetSessionStatus` for that window first. Sweep precedence when several conditions fire at once is lease-expiry, then faulted, then detach-grace.
 #### Reconnect and replay
 A client that drops mid-stream reconnects by re-issuing `StreamEvents` with `StreamEventsRequest.after_worker_sequence` set to the last `worker_sequence` it observed. A non-zero `after_worker_sequence` means *resume*; `0` means *fresh stream* and behaves exactly as a first-time subscribe — no replay, no sentinel.
 **Owner-scoped attach (security control).** `OpenSession` records the caller's API key id on the session as `GatewaySession.OwnerKeyId`. Every `StreamEvents` attach and reattach is owner-checked: `EventStreamService.StreamEventsAsync` compares the caller's key id (threaded from `GatewayGrpcAuthorizationInterceptor` via `MxAccessGatewayService.StreamEvents`) to the session owner and throws `SessionManagerException(PermissionDenied)` — surfaced as gRPC `PermissionDenied` — when they differ, before any subscriber is attached and before any replayed or live event is delivered. Possessing the `event` scope and knowing a session id is not enough. This is what makes the default-on detach-grace and replay-retention windows safe: without the check, any `event`-scoped key that learned a session id could attach to another key's retained session and receive its replayed and live data. A `null` owner (session opened with authentication disabled) matches only a `null` caller key.
 On a resume, `EventStreamService.StreamEventsAsync` attaches through `GatewaySession.AttachEventSubscriberWithReplay`, which calls `SessionEventDistributor.RegisterWithReplay`. That method snapshots the session's replay ring for events newer than `after_worker_sequence` **and** registers the live subscriber inside a single `_replayLock` critical section. This atomicity is what makes the replay→live handoff free of gaps and duplicates: the pump appends each event to the replay ring (under `_replayLock`) before fanning it to subscriber channels, so relative to that one critical section every event is either in the replay snapshot or fanned into the freshly-registered live channel — never both observably, never neither.
 The handoff is sealed by a watermark. `RegisterWithReplay` returns `LiveResumeSequence` (the highest replayed sequence, or `after_worker_sequence` when nothing was replayed); `EventStreamService` then filters the live channel to events strictly greater than that watermark. An event that was both included in the replay snapshot and — racing the registration — also written to the live channel has `worker_sequence <= LiveResumeSequence`, so the live filter drops it exactly once (no duplicate), while every newer event is delivered (no gap). The same per-item filter governs replayed and live events identically, so a constrained or resuming caller never sees a replayed event it could not have seen live.
@@ -143,6 +143,17 @@ session if the worker faults. Gated by `MxGateway:Alarms:Enabled` — see
 `docs/DesignDecisions.md` for why this reverses the v1 single-subscriber rule
 for the alarm subsystem.
 The monitor consumes its session's events as an **internal distributor
 subscriber** (`GatewaySession.AttachInternalEventSubscriber`), not by draining
 the worker event channel directly. The single worker event channel therefore has
 exactly one reader — the per-session `SessionEventDistributor` pump — which fans
 every event to both the dashboard mirror and the alarm feed; the alarm subscriber
 is internal (`isInternal: true`), so it is not counted against
 `MaxEventSubscribersPerSession` and a slow alarm reconcile can never fault the
 session. The worker event channel is single-reader and asserts it (a second
 `WorkerClient.ReadEventsAsync` consumer throws), so a regression cannot silently
 split the event stream between two drains.
 ### Alarm providers and failover
 The alarm feed has two providers, both implemented worker-side:
@@ -530,6 +541,7 @@ read, all of which contradict MXAccess semantics.
 - `element_data_type` that is `Raw` or `Unspecified`
 - an element `value` whose kind does not match `element_data_type`
 - `total_length` exceeds the gateway-configured maximum array length
  (`MxGateway:Events:MaxSparseArrayLength`, default 1,000,000)
 An empty `elements` list with a non-zero `total_length` is valid — it writes an
 all-defaults array of length `total_length` (explicit reset). A `sparse_array_value`
@@ -225,11 +225,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
            Task reconcileLoop = ReconcileLoopAsync(session.SessionId, linked.Token);
            try
            {
-                await foreach (WorkerEvent workerEvent in _sessionManager
+                // Consume mapped MxEvents through the session's single distributor pump (as an
-                    .ReadEventsAsync(session.SessionId, linked.Token)
+                // internal, non-counted subscriber) rather than opening a second raw drain of the
                // worker event channel — a second drain would split events with the dashboard
                // mirror pump and silently lose Acknowledge/mode-change transitions (GWC-01).
                await foreach (MxEvent mxEvent in _sessionManager
                    .ReadAlarmEventsAsync(session.SessionId, linked.Token)
                    .ConfigureAwait(false))
                {
                    MxEvent? mxEvent = workerEvent.Event;
                    if (mxEvent is { BodyCase: MxEvent.BodyOneofCase.OnAlarmTransition }
                        && mxEvent.OnAlarmTransition is not null)
                    {
@@ -27,4 +27,14 @@ public sealed class EventOptions
    ///     bounds the buffer).
    /// </summary>
    public double ReplayRetentionSeconds { get; init; } = 300;
    /// <summary>
    ///     Gets the maximum <c>total_length</c> a sparse-array write may declare before the
    ///     gateway rejects it with <c>InvalidArgument</c>, enforced in
    ///     <see cref="Sessions.SparseArrayExpander"/> before the full array is materialized.
    ///     Guards against a single write forcing a multi-GB allocation. The default of
    ///     1,000,000 elements is far above any realistic MXAccess array write yet well below
    ///     the frame-size ceiling.
    /// </summary>
    public int MaxSparseArrayLength { get; init; } = 1_000_000;
 }
@@ -181,6 +181,10 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
            options.DetachGraceSeconds,
            "MxGateway:Sessions:DetachGraceSeconds must be zero or greater (0 disables detach-grace retention).",
            builder);
        AddIfNegative(
            options.FaultedGraceSeconds,
            "MxGateway:Sessions:FaultedGraceSeconds must be zero or greater (0 reaps a faulted session on the next sweep).",
            builder);
        AddIfNegative(
            options.WorkerReadyWaitTimeoutMs,
            "MxGateway:Sessions:WorkerReadyWaitTimeoutMs must be greater than or equal to zero.",
@@ -214,6 +218,10 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
        builder.RequireThat(
            options.ReplayRetentionSeconds >= 0,
            "MxGateway:Events:ReplayRetentionSeconds must be greater than or equal to zero.");
        builder.RequireThat(
            options.MaxSparseArrayLength >= 1 && options.MaxSparseArrayLength <= Array.MaxLength,
            $"MxGateway:Events:MaxSparseArrayLength must be between 1 and {Array.MaxLength}.");
    }
    private static void ValidateDashboard(DashboardOptions options, ValidationBuilder builder)
@@ -45,6 +45,18 @@ public sealed class SessionOptions
    /// </remarks>
    public int DetachGraceSeconds { get; init; } = 30;
    /// <summary>
    ///     Gets the grace period, in seconds, that a faulted session is retained before the
    ///     lease monitor reaps it (killing its worker and freeing the session slot). A faulted
    ///     session is otherwise permanently unusable — every command fails the readiness check —
    ///     yet without this sweep it pins a session slot and a live x86 worker until its normal
    ///     lease expires (up to <see cref="DefaultLeaseSeconds"/>). A value of <c>0</c> (the
    ///     default) reaps a faulted session on the next sweep cycle, bounding the blast radius; a
    ///     positive value keeps the faulted session observable via <c>GetSessionStatus</c> for the
    ///     grace window before it is reclaimed. Must be greater than or equal to zero.
    /// </summary>
    public int FaultedGraceSeconds { get; init; }
    /// <summary>
    /// Gets a value indicating whether multiple event subscribers are allowed per session.
    /// </summary>
@@ -48,6 +48,7 @@ public sealed class EventStreamService(
    /// </remarks>
    public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
        StreamEventsRequest request,
        string? callerKeyId,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        if (!sessionManager.TryGetSession(request.SessionId, out GatewaySession? session) || session is null)
@@ -57,6 +58,19 @@ public sealed class EventStreamService(
                $"Session {request.SessionId} was not found.");
        }
        // Owner-scoped attach (TST-02, security control): a session's event stream may be
        // attached or reattached ONLY by the API key that opened the session. The detach-grace
        // and fan-out retention windows are on by default, so without this check any event-scoped
        // key that learns a session id could attach to another key's retained session and receive
        // its replayed and live data. Ordinal comparison; null owner (session opened with no auth)
        // matches only a null caller key.
        if (!string.Equals(session.OwnerKeyId, callerKeyId, StringComparison.Ordinal))
        {
            throw new SessionManagerException(
                SessionManagerErrorCode.PermissionDenied,
                $"Session {request.SessionId} is owned by a different API key; event-stream attach is owner-scoped.");
        }
        // No `using` here — subscriber.Dispose() is called exactly once in the finally
        // block below, which also disposes the reader. A `using` declaration would add a
        // second Dispose on the same path and double-decrement the session subscriber count.
@@ -11,9 +11,16 @@ public interface IEventStreamService
    /// Streams events for the specified session to the caller.
    /// </summary>
    /// <param name="request">Request payload.</param>
    /// <param name="callerKeyId">
    ///     The API key id of the calling client, used to enforce that only the key that opened a
    ///     session may attach or reattach its event stream. <see langword="null"/> when the call
    ///     is unauthenticated (e.g. auth disabled), which only matches a session opened with no
    ///     owner key.
    /// </param>
    /// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
    /// <returns>The events emitted for the requested session.</returns>
    IAsyncEnumerable<MxEvent> StreamEventsAsync(
        StreamEventsRequest request,
        string? callerKeyId,
        CancellationToken cancellationToken);
 }
@@ -148,7 +148,7 @@ public sealed class MxAccessGatewayService(
        {
            requestValidator.ValidateStreamEvents(request);
            await foreach (MxEvent publicEvent in eventStreamService
-                .StreamEventsAsync(request, context.CancellationToken)
+                .StreamEventsAsync(request, identityAccessor.Current?.KeyId, context.CancellationToken)
                .WithCancellation(context.CancellationToken)
                .ConfigureAwait(false))
            {
@@ -931,6 +931,7 @@ public sealed class MxAccessGatewayService(
            SessionManagerErrorCode.SessionLimitExceeded => StatusCode.ResourceExhausted,
            SessionManagerErrorCode.OpenFailed => StatusCode.Unavailable,
            SessionManagerErrorCode.CloseFailed => StatusCode.Unavailable,
            SessionManagerErrorCode.PermissionDenied => StatusCode.PermissionDenied,
            _ => StatusCode.Unavailable,
        };
@@ -23,8 +23,10 @@ public sealed class GatewaySession
    private bool _closeStarted;
    private int _activeEventSubscriberCount;
    private readonly TimeSpan _detachGrace;
    private readonly TimeSpan _faultedGrace;
    private readonly TimeSpan _workerReadyWaitTimeout;
    private DateTimeOffset? _detachedAtUtc;
    private DateTimeOffset? _faultedAtUtc;
    // True once at least one external subscriber attached SUCCESSFULLY. Detach-grace's
    // "last subscriber dropped" stamp (see DetachEventSubscriber) is gated on this so a
    // FAILED first attach — which still runs the rollback DetachEventSubscriber from the
@@ -139,6 +141,14 @@ public sealed class GatewaySession
    ///     When <see langword="null"/> (legacy unit-construction paths that do not exercise Galaxy
    ///     metadata), addresses pass through unchanged.
    /// </param>
    /// <param name="faultedGrace">
    ///     Grace window kept after the session faults before the lease monitor reaps it. When the
    ///     window is positive the faulted session stays observable via <c>GetSessionStatus</c> for
    ///     that long before it is reclaimed; <see cref="TimeSpan.Zero"/> (the default) makes the
    ///     session reapable on the next sweep. The fault timestamp is stamped in
    ///     <see cref="MarkFaulted"/> using <paramref name="eventStreaming"/>'s clock so the timer
    ///     is unit-testable.
    /// </param>
    public GatewaySession(
        string sessionId,
        string backendName,
@@ -156,7 +166,8 @@ public sealed class GatewaySession
        SessionEventStreaming? eventStreaming = null,
        TimeSpan detachGrace = default,
        TimeSpan workerReadyWaitTimeout = default,
-        ArrayAddressNormalizer? addressNormalizer = null)
+        ArrayAddressNormalizer? addressNormalizer = null,
        TimeSpan faultedGrace = default)
    {
        if (string.IsNullOrWhiteSpace(sessionId))
        {
@@ -195,6 +206,7 @@ public sealed class GatewaySession
        _leaseExpiresAt = openedAt + leaseDuration;
        _eventStreaming = eventStreaming ?? SessionEventStreaming.Default;
        _detachGrace = detachGrace > TimeSpan.Zero ? detachGrace : TimeSpan.Zero;
        _faultedGrace = faultedGrace > TimeSpan.Zero ? faultedGrace : TimeSpan.Zero;
        _workerReadyWaitTimeout = workerReadyWaitTimeout > TimeSpan.Zero ? workerReadyWaitTimeout : TimeSpan.Zero;
        _addressNormalizer = addressNormalizer;
    }
@@ -522,6 +534,33 @@ public sealed class GatewaySession
        }
    }
    /// <summary>
    ///     Registers a gateway-owned <em>internal</em> (non-counted) distributor subscriber and
    ///     returns its lease. The lease's <see cref="IEventSubscriberLease.Reader"/> yields the
    ///     same mapped <see cref="MxEvent"/>s the single distributor pump fans to every
    ///     subscriber; disposing the lease unregisters it.
    /// </summary>
    /// <remarks>
    ///     Used by the central alarm monitor so it consumes events through the one distributor
    ///     pump instead of opening a second raw drain of the single worker event channel (which
    ///     would split events between the two readers). Mirrors the dashboard-mirror lease:
    ///     <c>isInternal: true</c> keeps this subscriber out of the
    ///     <c>MaxEventSubscribersPerSession</c> accounting and out of the single-subscriber
    ///     overflow-fault path, so a slow alarm reconcile can never fault the session — it only
    ///     disconnects this internal subscriber.
    /// </remarks>
    /// <returns>The internal subscriber's lease; dispose it to unregister.</returns>
    public IEventSubscriberLease AttachInternalEventSubscriber()
    {
        // Same sequence StartDashboardMirror uses: create the distributor (claiming the pump
        // start if we are first), register the internal subscriber BEFORE the pump starts so a
        // subscriber is always present at pump start, then start the pump if requested.
        SessionEventDistributor distributor = EnsureDistributorCreated(out bool startNow);
        IEventSubscriberLease lease = distributor.Register(isInternal: true);
        StartPumpIfRequested(distributor, startNow);
        return lease;
    }
    private static void StartPumpIfRequested(SessionEventDistributor distributor, bool startNow)
    {
        if (!startNow)
@@ -724,6 +763,11 @@ public sealed class GatewaySession
            _finalFault = reason;
            _state = SessionState.Faulted;
            // Stamp the fault time once, on the first fault, so the sweeper can apply
            // FaultedGraceSeconds. A subsequent MarkFaulted (already-faulted session) keeps the
            // original timestamp so the grace window is measured from the first fault.
            _faultedAtUtc ??= _eventStreaming.TimeProvider.GetUtcNow();
        }
    }
@@ -790,6 +834,24 @@ public sealed class GatewaySession
        }
    }
    /// <summary>
    ///     Determines whether a faulted session is now eligible for reaping by the lease monitor.
    ///     A faulted session is permanently unusable (every command fails the readiness check),
    ///     so the sweeper closes it exactly as it closes an expired lease — but no sooner than the
    ///     configured <c>FaultedGraceSeconds</c> after the fault, so a monitoring client can still
    ///     observe the fault before the slot is reclaimed. Always returns <see langword="false"/>
    ///     for a non-faulted session.
    /// </summary>
    /// <param name="now">Current timestamp for comparison.</param>
    /// <returns><see langword="true"/> if the session is faulted and past its fault-grace window; otherwise <see langword="false"/>.</returns>
    public bool IsFaultedReapable(DateTimeOffset now)
    {
        lock (_syncRoot)
        {
            return IsFaultedReapableCore(now);
        }
    }
    /// <summary>
    ///     Attaches an event subscriber and returns a lease whose
    ///     <see cref="IEventSubscriberLease.Reader"/> reads the fanned public
@@ -1053,12 +1115,13 @@ public sealed class GatewaySession
        _addressNormalizer?.Normalize(address) ?? address;
    // MXAccess writes replace the whole array; expand a sparse value in place so the worker only
-    // ever receives a whole-array MxValue. No-op for null or non-sparse values.
+    // ever receives a whole-array MxValue. No-op for null or non-sparse values. The configured
-    private static void ExpandValue(MxValue? value)
+    // MxGateway:Events:MaxSparseArrayLength cap is enforced before the full array is allocated.
    private void ExpandValue(MxValue? value)
    {
        if (value is not null)
        {
-            SparseArrayExpander.Expand(value);
+            SparseArrayExpander.Expand(value, _eventStreaming.EventOptions.MaxSparseArrayLength);
        }
    }
@@ -1571,7 +1634,7 @@ public sealed class GatewaySession
            // Re-verify eligibility atomically. If a subscriber reattached between the sweep's
            // eligibility check and this point, neither condition holds and we decline.
-            bool eligible = IsLeaseExpiredCore(now) || IsDetachGraceExpiredCore(now);
+            bool eligible = IsLeaseExpiredCore(now) || IsFaultedReapableCore(now) || IsDetachGraceExpiredCore(now);
            if (!eligible)
            {
                alreadyClosing = false;
@@ -1597,6 +1660,12 @@ public sealed class GatewaySession
            && _detachedAtUtc is not null
            && now - _detachedAtUtc.Value >= _detachGrace;
    private bool IsFaultedReapableCore(DateTimeOffset now)
        => _state is SessionState.Faulted
            && (_faultedGrace <= TimeSpan.Zero
                || _faultedAtUtc is null
                || now - _faultedAtUtc.Value >= _faultedGrace);
    // Final terminal transition; under _syncRoot to keep _state writes single-lock.
    // Closed is unconditionally terminal — TransitionTo refuses to overwrite it —
    // so we don't need to re-check the precondition here.
@@ -43,6 +43,18 @@ public interface ISessionManager
        string sessionId,
        CancellationToken cancellationToken);
    /// <summary>
    ///     Reads mapped events for the central alarm monitor by attaching an internal
    ///     (non-counted) distributor subscriber, so the alarm feed shares the one worker-event
    ///     pump instead of opening a second raw drain of the single worker event channel.
    /// </summary>
    /// <param name="sessionId">Identifier of the session.</param>
    /// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
    /// <returns>The mapped <see cref="MxEvent"/>s fanned by the session's distributor.</returns>
    IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
        string sessionId,
        CancellationToken cancellationToken);
    /// <summary>Closes a session and terminates its worker process.</summary>
    /// <param name="sessionId">Identifier of the session to close.</param>
    /// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
@@ -1,4 +1,5 @@
 using System.Diagnostics.CodeAnalysis;
 using System.Runtime.CompilerServices;
 using System.Security.Cryptography;
 using Google.Protobuf.WellKnownTypes;
 using Microsoft.Extensions.Logging;
@@ -18,6 +19,7 @@ public sealed class SessionManager : ISessionManager
    public const string GatewayShutdownReason = "gateway-shutdown";
    public const string LeaseExpiredReason = "lease-expired";
    public const string DetachGraceExpiredReason = "detach-grace-expired";
    public const string FaultedReason = "faulted-reaped";
    private readonly ISessionRegistry _registry;
    private readonly ISessionWorkerClientFactory _workerClientFactory;
@@ -189,6 +191,22 @@ public sealed class SessionManager : ISessionManager
        return session.ReadEventsAsync(cancellationToken);
    }
    /// <inheritdoc />
    public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
        string sessionId,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        GatewaySession session = GetRequiredSession(sessionId);
        using IEventSubscriberLease lease = session.AttachInternalEventSubscriber();
        await foreach (MxEvent mxEvent in lease.Reader
            .ReadAllAsync(cancellationToken)
            .ConfigureAwait(false))
        {
            yield return mxEvent;
        }
    }
    /// <inheritdoc />
    public async Task<SessionCloseResult> CloseSessionAsync(
        string sessionId,
@@ -259,19 +277,22 @@ public sealed class SessionManager : ISessionManager
        int closedCount = 0;
        foreach (GatewaySession session in _registry.Snapshot())
        {
-            // A session is swept when its normal lease has expired OR its detach-grace
+            // A session is swept when its normal lease has expired, it has FAULTED (a faulted
-            // retention window has elapsed (last external subscriber dropped and no client
+            // session is permanently unusable yet otherwise pins a session slot and a live x86
-            // reconnected within DetachGraceSeconds). The detach-grace close is the same
+            // worker until its DefaultLeaseSeconds lease expires), OR its detach-grace retention
-            // teardown as a lease-expiry close; only the reason differs so operators can tell
+            // window has elapsed (last external subscriber dropped and no client reconnected
-            // a short reconnect-window expiry from a long idle-lease expiry in logs/metrics.
+            // within DetachGraceSeconds). All three are the same teardown; only the reason differs
-            // Lease-expiry takes PRECEDENCE over detach-grace when both conditions fire
+            // so operators can tell a fault reap from a short reconnect-window expiry from a long
-            // simultaneously (reason will be lease-expired, not detach-grace-expired).
+            // idle-lease expiry in logs/metrics. Precedence when several fire simultaneously:
            // lease-expired, then faulted, then detach-grace.
            //
            // TOCTOU note: eligibility is re-verified atomically inside TryBeginCloseIfExpired
            // under _syncRoot, so a client that reattaches a subscriber between the check above
            // and the close call wins the race and the session is left open and usable.
            string? reason = session.IsLeaseExpired(now)
                ? LeaseExpiredReason
                : session.IsFaultedReapable(now)
                    ? FaultedReason
                    : session.IsDetachGraceExpired(now)
                        ? DetachGraceExpiredReason
                        : null;
@@ -457,7 +478,8 @@ public sealed class SessionManager : ISessionManager
            eventStreaming,
            TimeSpan.FromSeconds(Math.Max(0, _options.Sessions.DetachGraceSeconds)),
            TimeSpan.FromMilliseconds(Math.Max(0, _options.Sessions.WorkerReadyWaitTimeoutMs)),
-            _addressNormalizer);
+            _addressNormalizer,
            TimeSpan.FromSeconds(Math.Max(0, _options.Sessions.FaultedGraceSeconds)));
    }
    private static string CreateClientCorrelationId(
@@ -10,4 +10,11 @@ public enum SessionManagerErrorCode
    SessionLimitExceeded,
    OpenFailed,
    CloseFailed,
    /// <summary>
    ///     The caller is not permitted to perform the operation on this session — for example,
    ///     attaching a <c>StreamEvents</c> stream to a session opened by a different API key.
    ///     Maps to gRPC <c>PermissionDenied</c>.
    /// </summary>
    PermissionDenied,
 }
@@ -33,13 +33,20 @@ internal static class SparseArrayExpander
    /// a sparse array this is a no-op, so callers may invoke it unconditionally.
    /// </summary>
    /// <param name="value">The value to expand in place.</param>
    /// <param name="maxSparseArrayLength">
    /// The maximum <c>total_length</c> the sparse array may declare before the write is
    /// rejected, enforced before the full array is allocated (see
    /// <c>MxGateway:Events:MaxSparseArrayLength</c>). Defaults to <see cref="int.MaxValue"/>
    /// so the <see cref="Array.MaxLength"/> backstop is the only bound in test/unit-construction
    /// paths that do not thread the configured cap.
    /// </param>
    /// <exception cref="RpcException">
    /// <see cref="StatusCode.InvalidArgument"/> when the sparse payload is invalid: zero
-    /// total length, an index at or beyond the total length, a duplicate index, an
+    /// total length, a total length exceeding <paramref name="maxSparseArrayLength"/>, an index
-    /// unsupported element type, or an element value whose kind does not match the declared
+    /// at or beyond the total length, a duplicate index, an unsupported element type, or an
-    /// element type.
+    /// element value whose kind does not match the declared element type.
    /// </exception>
-    public static void Expand(MxValue value)
+    public static void Expand(MxValue value, int maxSparseArrayLength = int.MaxValue)
    {
        ArgumentNullException.ThrowIfNull(value);
@@ -62,6 +69,12 @@ internal static class SparseArrayExpander
            throw Invalid($"Sparse array element_data_type '{elementType}' is not a supported scalar element type.");
        }
        if (totalLength > (uint)maxSparseArrayLength)
        {
            throw Invalid(
                $"Sparse array total_length {totalLength} exceeds the configured maximum {maxSparseArrayLength} (MxGateway:Events:MaxSparseArrayLength).");
        }
        if (totalLength > (uint)Array.MaxLength)
        {
            throw Invalid(
@@ -32,6 +32,7 @@ public sealed class WorkerClient : IWorkerClient
    private DateTimeOffset _lastHeartbeatAt;
    private int? _processId;
    private int _eventQueueDepth;
    private int _eventsReaderClaimed;
    private Task? _readLoopTask;
    private Task? _writeLoopTask;
    private Task? _heartbeatLoopTask;
@@ -70,7 +71,13 @@ public sealed class WorkerClient : IWorkerClient
        _events = Channel.CreateBounded<WorkerEvent>(
            new BoundedChannelOptions(_options.EventChannelCapacity)
            {
-                SingleReader = false,
+                // The worker event channel has exactly ONE consumer: the per-session
                // SessionEventDistributor pump. The alarm monitor and dashboard mirror both
                // attach to the distributor rather than draining this channel directly, so a
                // second concurrent reader would silently split events between the two
                // enumerators. SingleReader=true asserts that invariant; ReadEventsAsync adds a
                // claimed-once guard so a regression fails loudly instead of losing events.
                SingleReader = true,
                SingleWriter = true,
                FullMode = BoundedChannelFullMode.Wait,
                AllowSynchronousContinuations = false,
@@ -224,7 +231,24 @@ public sealed class WorkerClient : IWorkerClient
    }
    /// <inheritdoc />
-    public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+    public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(CancellationToken cancellationToken)
    {
        // The event channel is SingleReader: only one enumerator may ever drain it, otherwise
        // the two readers would each receive a random subset of events. Claim the reader at CALL
        // time (not lazily on first MoveNext) and fail loudly on a second consumer rather than
        // silently splitting the stream (see GWC-01). The distributor pump is the only intended
        // caller; the alarm monitor and dashboard mirror attach to the distributor instead.
        if (Interlocked.CompareExchange(ref _eventsReaderClaimed, 1, 0) != 0)
        {
            throw new InvalidOperationException(
                "WorkerClient.ReadEventsAsync was already claimed by another consumer. The worker event "
                + "channel is single-reader; attach to the SessionEventDistributor instead of draining it twice.");
        }
        return ReadEventsCoreAsync(cancellationToken);
    }
    private async IAsyncEnumerable<WorkerEvent> ReadEventsCoreAsync(
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
@@ -470,6 +470,20 @@ public sealed class AlarmFailoverEndToEndTests
            }
        }
        /// <inheritdoc />
        public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
            string sessionId,
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
            {
                if (workerEvent.Event is not null)
                {
                    yield return workerEvent.Event;
                }
            }
        }
        /// <inheritdoc />
        public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
        {
@@ -783,6 +783,20 @@ public sealed class GatewayAlarmMonitorProviderModeTests
            }
        }
        /// <inheritdoc />
        public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
            string sessionId,
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
            {
                if (workerEvent.Event is not null)
                {
                    yield return workerEvent.Event;
                }
            }
        }
        /// <inheritdoc />
        public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
        {
@@ -520,4 +520,32 @@ public sealed class GatewayOptionsValidatorTests
        ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
        Assert.True(result.Succeeded);
    }
    /// <summary>Verifies a <see cref="EventOptions.MaxSparseArrayLength"/> below one fails validation.</summary>
    /// <param name="value">Sparse-array cap under test.</param>
    [Theory]
    [InlineData(0)]
    [InlineData(-1)]
    public void Validate_Fails_WhenMaxSparseArrayLengthBelowOne(int value)
    {
        GatewayOptions options = CloneWithEvents(
            ValidOptions(),
            new EventOptions { MaxSparseArrayLength = value });
        ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
        Assert.True(result.Failed);
        Assert.Contains(
            result.Failures!,
            f => f.Contains("MxGateway:Events:MaxSparseArrayLength"));
    }
    /// <summary>Verifies a positive <see cref="EventOptions.MaxSparseArrayLength"/> within range passes validation.</summary>
    [Fact]
    public void Validate_Succeeds_WhenMaxSparseArrayLengthWithinRange()
    {
        GatewayOptions options = CloneWithEvents(
            ValidOptions(),
            new EventOptions { MaxSparseArrayLength = 1 });
        ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
        Assert.True(result.Succeeded);
    }
 }
@@ -281,6 +281,14 @@ public sealed class DashboardSessionAdminServiceTests
            throw new NotSupportedException();
        }
        /// <inheritdoc />
        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            throw new NotSupportedException();
        }
        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(
            string sessionId,
@@ -38,6 +38,58 @@ public sealed class EventStreamServiceTests
        Assert.Equal(1, metrics.GetSnapshot().StreamDisconnects);
    }
    /// <summary>
    ///     TST-02 (owner-scoped attach): the API key that opened a session may attach its event
    ///     stream — the caller key equals the session owner, so streaming proceeds normally.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenCallerKeyMatchesOwner_Streams()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient, ownerKeyId: "key-owner");
        EventStreamService service = CreateService(new FakeSessionManager(session));
        workerClient.Events.Add(CreateWorkerEvent(sequence: 5, MxEventFamily.OnDataChange));
        workerClient.CompleteAfterConfiguredEvents = true;
        List<MxEvent> events = [];
        await foreach (MxEvent mxEvent in service
            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: "key-owner", CancellationToken.None)
            .WithCancellation(CancellationToken.None))
        {
            events.Add(mxEvent);
        }
        Assert.Equal([5UL], events.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
    }
    /// <summary>
    ///     TST-02 (owner-scoped attach, security control): a caller whose API key differs from
    ///     the key that opened the session is rejected with a <see cref="SessionManagerErrorCode.PermissionDenied"/>
    ///     fault before any events are streamed — closing the reconnect/fan-out trust-boundary hole.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenCallerKeyDiffersFromOwner_ThrowsPermissionDenied()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient, ownerKeyId: "key-owner");
        EventStreamService service = CreateService(new FakeSessionManager(session));
        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(async () =>
        {
            await foreach (MxEvent _ in service
                .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: "key-intruder", CancellationToken.None)
                .WithCancellation(CancellationToken.None))
            {
                // No event should be yielded — the owner check runs before the first attach.
            }
        });
        Assert.Equal(SessionManagerErrorCode.PermissionDenied, exception.ErrorCode);
        Assert.Equal(0, session.ActiveEventSubscriberCount);
    }
    /// <summary>Verifies that a second event subscriber is rejected when one is already active.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
@@ -48,13 +100,13 @@ public sealed class EventStreamServiceTests
        EventStreamService service = CreateService(new FakeSessionManager(session));
        using CancellationTokenSource firstSubscriberCancellation = new();
        await using IAsyncEnumerator<MxEvent> firstSubscriber = service
-            .StreamEventsAsync(CreateRequest(session.SessionId), firstSubscriberCancellation.Token)
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, firstSubscriberCancellation.Token)
            .GetAsyncEnumerator(firstSubscriberCancellation.Token);
        Task<bool> firstMoveTask = firstSubscriber.MoveNextAsync().AsTask();
        await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 1);
        await using IAsyncEnumerator<MxEvent> secondSubscriber = service
-            .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
@@ -78,7 +130,7 @@ public sealed class EventStreamServiceTests
        EventStreamService service = CreateService(new FakeSessionManager(session));
        using CancellationTokenSource cancellationTokenSource = new();
        await using IAsyncEnumerator<MxEvent> subscriber = service
-            .StreamEventsAsync(CreateRequest(session.SessionId), cancellationTokenSource.Token)
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, cancellationTokenSource.Token)
            .GetAsyncEnumerator(cancellationTokenSource.Token);
        Task<bool> moveTask = subscriber.MoveNextAsync().AsTask();
@@ -108,7 +160,7 @@ public sealed class EventStreamServiceTests
        workerClient.Events.Add(CreateWorkerEvent(sequence: 3, MxEventFamily.OnDataChange));
        workerClient.CompleteAfterConfiguredEvents = true;
        await using IAsyncEnumerator<MxEvent> subscriber = service
-            .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        Assert.True(await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
@@ -142,10 +194,10 @@ public sealed class EventStreamServiceTests
        firstWorkerClient.CompleteAfterConfiguredEvents = true;
        secondWorkerClient.CompleteAfterConfiguredEvents = true;
        await using IAsyncEnumerator<MxEvent> firstSubscriber = service
-            .StreamEventsAsync(CreateRequest(firstSession.SessionId), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(firstSession.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        await using IAsyncEnumerator<MxEvent> secondSubscriber = service
-            .StreamEventsAsync(CreateRequest(secondSession.SessionId), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(secondSession.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        Assert.True(await firstSubscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
@@ -192,7 +244,7 @@ public sealed class EventStreamServiceTests
        workerClient.CompleteAfterConfiguredEvents = true;
        await using IAsyncEnumerator<MxEvent> subscriber = service
-            .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        // The pump fans 50 events into a subscriber channel with capacity 1 faster than this
@@ -246,7 +298,7 @@ public sealed class EventStreamServiceTests
        workerClient.CompleteAfterConfiguredEvents = true;
        await using IAsyncEnumerator<MxEvent> subscriber = service
-            .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
@@ -298,7 +350,7 @@ public sealed class EventStreamServiceTests
        using GatewayMetrics metrics = new();
        EventStreamService service = CreateService(new FakeSessionManager(session), metrics);
        await using IAsyncEnumerator<MxEvent> subscriber = service
-            .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
@@ -333,7 +385,7 @@ public sealed class EventStreamServiceTests
        // Resume after sequence 2: retained window [1..5] covers it — replay 3,4,5 then live.
        await using IAsyncEnumerator<MxEvent> resume = service
-            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        MxEvent r3 = await ReadNextAsync(resume);
@@ -374,7 +426,7 @@ public sealed class EventStreamServiceTests
        // Resume after 1: events 1,2 are below the oldest retained (3) and were evicted, so
        // they are unrecoverable => sentinel first, then the retained tail 3,4,5, then live.
        await using IAsyncEnumerator<MxEvent> realResume = service
-            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 1), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 1), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        MxEvent sentinel = await ReadNextAsync(realResume);
@@ -419,7 +471,7 @@ public sealed class EventStreamServiceTests
        // Resume after 2: replay 3,4 then live 5,6,7. Collect across the boundary and assert
        // the full sequence is contiguous with no duplicate and no skip.
        await using IAsyncEnumerator<MxEvent> resume = service
-            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        List<ulong> collected = [];
@@ -461,7 +513,7 @@ public sealed class EventStreamServiceTests
        // reads must be exactly 4 then 5 (no sentinel, no <=3 event); a live tag confirms the
        // stream resumed live strictly after 5.
        await using IAsyncEnumerator<MxEvent> resume = service
-            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 3), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 3), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        MxEvent first = await ReadNextAsync(resume);
@@ -512,7 +564,7 @@ public sealed class EventStreamServiceTests
        int expectedCount)
    {
        await using IAsyncEnumerator<MxEvent> primer = service
-            .StreamEventsAsync(CreateRequest(sessionId), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(sessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        for (int i = 0; i < expectedCount; i++)
        {
@@ -551,7 +603,7 @@ public sealed class EventStreamServiceTests
    {
        List<MxEvent> events = [];
        await foreach (MxEvent mxEvent in service
-            .StreamEventsAsync(CreateRequest(sessionId), CancellationToken.None)
+            .StreamEventsAsync(CreateRequest(sessionId), callerKeyId: null, CancellationToken.None)
            .WithCancellation(CancellationToken.None))
        {
            events.Add(mxEvent);
@@ -575,7 +627,8 @@ public sealed class EventStreamServiceTests
        int queueCapacity = 8,
        GatewayMetrics? metrics = null,
        EventBackpressurePolicy backpressurePolicy = EventBackpressurePolicy.FailFast,
-        int replayBufferCapacity = 1024)
+        int replayBufferCapacity = 1024,
        string? ownerKeyId = null)
    {
        // The per-subscriber overflow policy now lives in the session's
        // SessionEventDistributor, so the session must share the same metrics sink and
@@ -587,7 +640,7 @@ public sealed class EventStreamServiceTests
            "pipe",
            "nonce",
            "client",
-            ownerKeyId: null,
+            ownerKeyId: ownerKeyId,
            "client-session",
            "client-correlation",
            TimeSpan.FromSeconds(30),
@@ -702,6 +755,14 @@ public sealed class EventStreamServiceTests
            return _sessions[sessionId].ReadEventsAsync(cancellationToken);
        }
        /// <inheritdoc />
        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            throw new NotSupportedException();
        }
        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(
            string sessionId,
@@ -948,6 +948,14 @@ public sealed class MxAccessGatewayServiceConstraintTests
            }
        }
        /// <inheritdoc />
        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            throw new NotSupportedException();
        }
        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(
            string sessionId,
@@ -994,6 +1002,7 @@ public sealed class MxAccessGatewayServiceConstraintTests
        /// <inheritdoc />
        public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
            StreamEventsRequest request,
            string? callerKeyId,
            [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
        {
            foreach (WorkerEvent ev in sessionManager.Events)
@@ -616,6 +616,14 @@ public sealed class MxAccessGatewayServiceTests
            }
        }
        /// <inheritdoc />
        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            throw new NotSupportedException();
        }
        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(
            string sessionId,
@@ -653,6 +661,7 @@ public sealed class MxAccessGatewayServiceTests
        /// <inheritdoc />
        public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
            StreamEventsRequest request,
            string? callerKeyId,
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            sessionManager.RecordReadEventsSessionId(request.SessionId);
@@ -88,7 +88,7 @@ public sealed class GatewaySessionDashboardMirrorTests
        Task grpcReader = Task.Run(async () =>
        {
            await foreach (MxEvent mxEvent in service
-                .StreamEventsAsync(new StreamEventsRequest { SessionId = session.SessionId }, CancellationToken.None)
+                .StreamEventsAsync(new StreamEventsRequest { SessionId = session.SessionId }, callerKeyId: null, CancellationToken.None)
                .WithCancellation(CancellationToken.None))
            {
                grpcEvents.Add(mxEvent);
@@ -107,6 +107,65 @@ public sealed class GatewaySessionDashboardMirrorTests
        Assert.Equal([1UL, 2UL, 3UL], broadcaster.Captures.Select(capture => capture.MxEvent.WorkerSequence).ToArray());
    }
    /// <summary>
    ///     GWC-01 regression: with the internal dashboard mirror active, a second internal
    ///     subscriber (the alarm monitor's feed, attached via
    ///     <see cref="GatewaySession.AttachInternalEventSubscriber"/>) receives EVERY event —
    ///     including the alarm <c>Acknowledge</c> transition — rather than the two consumers
    ///     each getting a random half of the single worker channel. Before the fix the alarm
    ///     monitor drained the worker channel directly, so with the dashboard mirror pump also
    ///     draining it the two split the stream and Acknowledge transitions were silently lost.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task InternalAlarmSubscriber_AndDashboardMirror_BothReceiveEveryEvent()
    {
        FakeWorkerClient workerClient = new();
        workerClient.Events.Add(CreateWorkerEvent(1, MxEventFamily.OnDataChange));
        workerClient.Events.Add(CreateAlarmTransitionEvent(2, AlarmTransitionKind.Raise));
        workerClient.Events.Add(CreateAlarmTransitionEvent(3, AlarmTransitionKind.Acknowledge));
        workerClient.Events.Add(CreateAlarmTransitionEvent(4, AlarmTransitionKind.Clear));
        workerClient.CompleteAfterConfiguredEvents = true;
        // Hold the finite stream until BOTH internal subscribers (dashboard mirror + alarm feed)
        // are registered so neither misses an event before the pump drains.
        workerClient.HoldEventsUntilReleased();
        RecordingDashboardEventBroadcaster broadcaster = new();
        await using GatewaySession session = CreateSession(workerClient, broadcaster);
        session.AttachWorkerClient(workerClient);
        // MarkReady registers the internal dashboard subscriber and starts the (gated) pump.
        session.MarkReady();
        // Attach the alarm monitor's internal subscriber and drain it concurrently. Registered
        // BEFORE the stream is released, so it is present at pump start alongside the dashboard.
        using IEventSubscriberLease alarmLease = session.AttachInternalEventSubscriber();
        List<MxEvent> alarmEvents = [];
        Task alarmReader = Task.Run(async () =>
        {
            await foreach (MxEvent mxEvent in alarmLease.Reader.ReadAllAsync(CancellationToken.None))
            {
                alarmEvents.Add(mxEvent);
            }
        });
        workerClient.ReleaseEvents();
        await WaitUntilAsync(() => broadcaster.Captures.Count == 4 && alarmEvents.Count == 4);
        await alarmReader.WaitAsync(TestTimeout);
        // Both internal consumers see the full, identical, ordered stream — no splitting.
        Assert.Equal([1UL, 2UL, 3UL, 4UL], broadcaster.Captures.Select(capture => capture.MxEvent.WorkerSequence).ToArray());
        Assert.Equal([1UL, 2UL, 3UL, 4UL], alarmEvents.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
        // The Acknowledge transition specifically reaches the alarm feed (the transition the
        // pre-fix split silently dropped, leaving clients showing unacked alarms indefinitely).
        Assert.Contains(
            alarmEvents,
            mxEvent => mxEvent.BodyCase == MxEvent.BodyOneofCase.OnAlarmTransition
                && mxEvent.OnAlarmTransition.TransitionKind == AlarmTransitionKind.Acknowledge);
    }
    /// <summary>
    ///     Hazard guard: starting the pump at Ready with a fast-completing worker stream
    ///     and zero subscribers used to drain into nothing and leave a later subscriber hanging.
@@ -222,6 +281,19 @@ public sealed class GatewaySessionDashboardMirrorTests
        return new WorkerEvent { Event = mxEvent };
    }
    private static WorkerEvent CreateAlarmTransitionEvent(ulong sequence, AlarmTransitionKind kind)
    {
        MxEvent mxEvent = new()
        {
            SessionId = "session-dashboard-mirror",
            Family = MxEventFamily.OnAlarmTransition,
            WorkerSequence = sequence,
            OnAlarmTransition = new OnAlarmTransitionEvent { TransitionKind = kind },
        };
        return new WorkerEvent { Event = mxEvent };
    }
    private static async Task WaitUntilAsync(Func<bool> predicate, [CallerArgumentExpression(nameof(predicate))] string? condition = null)
    {
        using CancellationTokenSource cancellationTokenSource = new(TestTimeout);
@@ -280,6 +352,11 @@ public sealed class GatewaySessionDashboardMirrorTests
            string sessionId,
            CancellationToken cancellationToken) => session.ReadEventsAsync(cancellationToken);
        /// <inheritdoc />
        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
            string sessionId,
            CancellationToken cancellationToken) => throw new NotSupportedException();
        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(
            string sessionId,
@@ -975,6 +975,68 @@ public sealed class SessionManagerTests
        Assert.Equal(0, workerClient.ShutdownCount);
    }
    /// <summary>
    ///     A faulted session is reaped by the lease sweep (default <c>FaultedGraceSeconds=0</c>)
    ///     even though its normal lease is still far in the future, tearing down its worker and
    ///     freeing the slot, while a healthy leased session in the same manager is untouched.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task CloseExpiredLeasesAsync_ReapsFaultedSession()
    {
        FakeWorkerClient faultedClient = new();
        FakeWorkerClient healthyClient = new();
        QueueingSessionWorkerClientFactory factory = new(faultedClient, healthyClient);
        SessionManager manager = CreateManager(factory);
        GatewaySession faultedSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
        GatewaySession healthySession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None);
        DateTimeOffset now = DateTimeOffset.UtcNow;
        // Both leases are far in the future, so only the fault can reap the faulted session.
        faultedSession.ExtendLease(now.AddMinutes(30));
        healthySession.ExtendLease(now.AddMinutes(30));
        faultedSession.MarkFaulted("test fault");
        int closedCount = await manager.CloseExpiredLeasesAsync(now, CancellationToken.None);
        Assert.Equal(1, closedCount);
        Assert.Equal(SessionState.Closed, faultedSession.State);
        Assert.Equal(1, faultedClient.ShutdownCount);
        Assert.Equal(SessionState.Ready, healthySession.State);
        Assert.Equal(0, healthyClient.ShutdownCount);
    }
    /// <summary>
    ///     With a positive <c>FaultedGraceSeconds</c>, a faulted session stays observable until
    ///     the grace window elapses, then the next sweep reaps it.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task CloseExpiredLeasesAsync_RespectsFaultedGraceWindow()
    {
        FakeWorkerClient workerClient = new();
        FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
        SessionManager manager = CreateManager(
            new FakeSessionWorkerClientFactory(workerClient),
            options: CreateOptions(defaultLeaseSeconds: 1800, faultedGraceSeconds: 30),
            timeProvider: clock);
        GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
        session.MarkFaulted("test fault");
        // Before the grace window elapses: the faulted session is retained (still observable).
        clock.Advance(TimeSpan.FromSeconds(29));
        int closedBefore = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
        Assert.Equal(0, closedBefore);
        Assert.Equal(SessionState.Faulted, session.State);
        // After the grace window elapses: the sweep reaps it.
        clock.Advance(TimeSpan.FromSeconds(1));
        int closedAfter = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
        Assert.Equal(1, closedAfter);
        Assert.Equal(SessionState.Closed, session.State);
        Assert.Equal(1, workerClient.ShutdownCount);
    }
    /// <summary>Verifies that shutdown closes all registered sessions.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
@@ -1023,7 +1085,8 @@ public sealed class SessionManagerTests
        int maxSessions = 64,
        int defaultLeaseSeconds = 1800,
        int detachGraceSeconds = 0,
-        int workerReadyWaitTimeoutMs = 0)
+        int workerReadyWaitTimeoutMs = 0,
        int faultedGraceSeconds = 0)
    {
        return new GatewayOptions
        {
@@ -1034,6 +1097,7 @@ public sealed class SessionManagerTests
                DefaultLeaseSeconds = defaultLeaseSeconds,
                DetachGraceSeconds = detachGraceSeconds,
                WorkerReadyWaitTimeoutMs = workerReadyWaitTimeoutMs,
                FaultedGraceSeconds = faultedGraceSeconds,
            },
            Worker = new WorkerOptions
            {
@@ -222,4 +222,35 @@ public sealed class SparseArrayExpanderTests
        RpcException ex = Assert.Throws<RpcException>(() => SparseArrayExpander.Expand(value));
        Assert.Equal(StatusCode.InvalidArgument, ex.StatusCode);
    }
    /// <summary>Verifies that a total length above the configured cap throws <see cref="StatusCode.InvalidArgument"/> before the full array is allocated.</summary>
    [Fact]
    public void Expand_TotalLengthExceedsConfiguredCap_ThrowsBeforeAllocation()
    {
        // A total_length that would force a huge allocation, but well below Array.MaxLength,
        // so only the configured cap can reject it (the Array.MaxLength backstop would not).
        MxValue value = SparseValue(MxDataType.Integer, 500_000_000u);
        RpcException ex = Assert.Throws<RpcException>(() => SparseArrayExpander.Expand(value, maxSparseArrayLength: 1_000_000));
        Assert.Equal(StatusCode.InvalidArgument, ex.StatusCode);
        Assert.Contains("MaxSparseArrayLength", ex.Status.Detail, StringComparison.Ordinal);
        // The value must be untouched — expansion (allocation) never ran.
        Assert.Equal(MxValue.KindOneofCase.SparseArrayValue, value.KindCase);
    }
    /// <summary>Verifies that a total length at or below the configured cap still expands normally.</summary>
    [Fact]
    public void Expand_TotalLengthAtConfiguredCap_Expands()
    {
        MxValue value = SparseValue(
            MxDataType.Integer,
            4,
            (1, new MxValue { Int32Value = 7 }));
        SparseArrayExpander.Expand(value, maxSparseArrayLength: 4);
        Assert.Equal(MxValue.KindOneofCase.ArrayValue, value.KindCase);
        Assert.Equal(new[] { 0, 7, 0, 0 }, value.ArrayValue.Int32Values.Values);
    }
 }
@@ -122,6 +122,27 @@ public sealed class WorkerClientTests
        Assert.Equal(MxEventFamily.OperationComplete, events.Current.Event.Family);
    }
    /// <summary>
    ///     The worker event channel is single-reader: a second <see cref="WorkerClient.ReadEventsAsync"/>
    ///     enumerator must throw rather than silently split events between two consumers (GWC-01).
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadEventsAsync_SecondEnumerator_Throws()
    {
        await using PipePair pipePair = await PipePair.CreateAsync();
        await using WorkerClient client = CreateClient(pipePair);
        await CompleteHandshakeAsync(client, pipePair);
        using CancellationTokenSource cancellationTokenSource = new(TestTimeout);
        // First call claims the single reader.
        IAsyncEnumerable<WorkerEvent> first = client.ReadEventsAsync(cancellationTokenSource.Token);
        await using IAsyncEnumerator<WorkerEvent> firstEnumerator = first.GetAsyncEnumerator(cancellationTokenSource.Token);
        // A second call must fail loudly at call time rather than racing the first for events.
        Assert.Throws<InvalidOperationException>(() => client.ReadEventsAsync(cancellationTokenSource.Token));
    }
    /// <summary>Verifies that the read loop faults the client when the event queue overflows.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
@@ -477,6 +477,14 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests
            return AsyncEnumerable.Empty<WorkerEvent>();
        }
        /// <inheritdoc />
        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            return AsyncEnumerable.Empty<MxEvent>();
        }
        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(
            string sessionId,
@@ -515,6 +523,7 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests
        /// <inheritdoc />
        public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
            StreamEventsRequest request,
            string? callerKeyId,
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            await Task.CompletedTask;
@@ -704,6 +704,102 @@ public sealed class WorkerPipeSessionTests
        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
    }
    /// <summary>
    ///     WRK-01 regression: a long in-flight STA command that keeps pumping
    ///     must NOT self-fault as <c>StaHung</c>, and its reply must still be
    ///     delivered. The real fix makes <c>StaRuntime.PumpPendingMessages</c>
    ///     refresh <c>LastActivityUtc</c> on every wait iteration, so a healthy
    ///     <c>ReadBulk</c> holding the STA far longer than
    ///     <c>HeartbeatStuckCeiling</c> (75 s in production) keeps its activity
    ///     timestamp fresh. This test compresses the clock — a 100 ms ceiling
    ///     with a command in flight across a window many multiples longer — and
    ///     models the pump refresh by continuously advancing the snapshot's
    ///     <c>LastStaActivityUtc</c> while the command blocks. Contrast
    ///     <see cref="RunAsync_WhenStaActivityIsStaleBeyondCeilingWithCommandInFlight_WritesWatchdogFault"/>,
    ///     where a frozen timestamp beyond the ceiling correctly faults; here
    ///     the refreshed timestamp must keep the fault suppressed and let the
    ///     reply through the <c>Ready</c>-state gate.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task RunAsync_LongInFlightCommandThatKeepsPumping_DoesNotFaultAndDeliversReply()
    {
        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(20));
        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
        FakeRuntimeSession runtime = new()
        {
            BlockDispatch = true,
        };
        WorkerPipeSession session = CreatePipeSession(
            pipePair.WorkerStream,
            runtime,
            new WorkerPipeSessionOptions
            {
                HeartbeatInterval = TimeSpan.FromMilliseconds(20),
                HeartbeatGrace = TimeSpan.FromMilliseconds(50),
                HeartbeatStuckCeiling = TimeSpan.FromMilliseconds(100),
            });
        Task runTask = session.RunAsync(cancellation.Token);
        await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
        // Kick off the long command; it blocks in DispatchAsync until released,
        // so its correlation id stays in flight in the heartbeat snapshot.
        await pipePair.GatewayWriter
            .WriteAsync(CreateCommandEnvelope("long-bulk-read"), cancellation.Token);
        Assert.True(
            runtime.DispatchStarted.Wait(TimeSpan.FromSeconds(5)),
            "The long command must reach the runtime and begin dispatch.");
        // Model the pump refreshing STA activity on each wait iteration: keep
        // the snapshot's LastStaActivityUtc current while the command is in
        // flight.
        using CancellationTokenSource pumpRefresh = new();
        Task refreshLoop = Task.Run(
            async () =>
            {
                while (!pumpRefresh.IsCancellationRequested)
                {
                    runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
                        DateTimeOffset.UtcNow,
                        pendingCommandCount: 1,
                        outboundEventQueueDepth: 0,
                        lastEventSequence: 0,
                        currentCommandCorrelationId: "long-bulk-read"));
                    await Task.Delay(TimeSpan.FromMilliseconds(20)).ConfigureAwait(false);
                }
            });
        // Inspect a bounded number of frames over a window many multiples of the
        // 100 ms ceiling (at least 30 heartbeats at 20 ms ~ 600 ms). None may be
        // a WorkerFault while activity is continuously refreshed.
        const int framesToInspect = 30;
        for (int index = 0; index < framesToInspect; index++)
        {
            WorkerEnvelope envelope = await pipePair.GatewayReader
                .ReadAsync(cancellation.Token);
            Assert.NotEqual(
                WorkerEnvelope.BodyOneofCase.WorkerFault,
                envelope.BodyCase);
        }
        // Stop refreshing and release the command; its reply must be delivered
        // because the session never faulted (state stayed Ready).
        pumpRefresh.Cancel();
        await refreshLoop;
        runtime.ReleaseDispatch();
        WorkerEnvelope reply = await ReadUntilAsync(
            pipePair.GatewayReader,
            WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
            envelope => envelope.CorrelationId == "long-bulk-read",
            cancellation.Token);
        Assert.Equal(
            ProtocolStatusCode.Ok,
            reply.WorkerCommandReply.Reply.ProtocolStatus.Code);
        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
    }
    /// <summary>
    ///     <c>RunAsync</c> must throw a diagnostic exception if the
    ///     runtime-session factory returns null, rather than deferring the
@@ -85,6 +85,34 @@ public sealed class StaRuntimeTests
        Assert.True(updated);
    }
    /// <summary>
    ///     Verifies that <see cref="StaRuntime.PumpPendingMessages"/> refreshes
    ///     <see cref="StaRuntime.LastActivityUtc"/>. A long synchronous STA
    ///     command (for example <c>ReadBulk</c> waiting <c>timeout_ms</c> for
    ///     the first <c>OnDataChange</c>) invokes the pump step on every wait
    ///     iteration while it legitimately holds the STA thread; refreshing
    ///     activity here keeps the watchdog from mistaking a busy STA for a hung
    ///     one (WRK-01). The runtime is deliberately left unstarted so the only
    ///     source of activity is the pump call under test, not the idle loop.
    /// </summary>
    [Fact]
    public void PumpPendingMessages_RefreshesLastActivity()
    {
        RecordingComApartmentInitializer initializer = new();
        using StaRuntime runtime = CreateRuntime(initializer);
        DateTimeOffset before = runtime.LastActivityUtc;
        bool refreshed = SpinWait.SpinUntil(
            () =>
            {
                runtime.PumpPendingMessages();
                return runtime.LastActivityUtc > before;
            },
            TimeSpan.FromSeconds(2));
        Assert.True(refreshed, "PumpPendingMessages must advance LastActivityUtc.");
    }
    /// <summary>Verifies that InvokeAsync faults the returned task when a command raises an exception without stopping the runtime.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
@@ -80,14 +80,24 @@ public sealed class StaRuntime : IDisposable
    public bool IsRunning => startedEvent.IsSet && !stoppedEvent.IsSet;
    /// <summary>
-    ///     Pumps any pending Windows messages on the calling thread. Intended
+    ///     Pumps any pending Windows messages on the calling thread and refreshes
-    ///     for commands that synchronously hold the STA (e.g. ReadBulk) and
+    ///     the STA activity timestamp. Intended for commands that synchronously
-    ///     must allow inbound MXAccess COM events to dispatch while they
+    ///     hold the STA (e.g. ReadBulk) and must allow inbound MXAccess COM events
-    ///     wait. Callers must already be on the STA; the method is otherwise
+    ///     to dispatch while they wait. Because a long-running command invokes this
-    ///     safe (PeekMessage simply finds no messages).
+    ///     on every wait iteration, refreshing activity here keeps a busy STA from
    ///     being mistaken for a hung one: a healthy command that keeps pumping stays
    ///     fresh past <c>HeartbeatStuckCeiling</c>, while a genuinely stuck STA (no
    ///     pumping) still accrues staleness and faults correctly. Callers must
    ///     already be on the STA; the method is otherwise safe (PeekMessage simply
    ///     finds no messages).
    /// </summary>
    /// <returns>The number of messages pumped.</returns>
-    public int PumpPendingMessages() => messagePump.PumpPendingMessages();
+    public int PumpPendingMessages()
    {
        int pumpedMessages = messagePump.PumpPendingMessages();
        MarkActivity();
        return pumpedMessages;
    }
    /// <summary>
    /// Starts the STA thread.