Implement worker register and unregister

Issue #14: implement event streaming and backpressure

docs/gateway-process-design.md

@@ -64,8 +64,8 @@ MxGateway.Server
   Configuration
   Grpc
     MxAccessGatewayService
-    RequestReplyMapper
-    EventMapper
+    MxAccessGrpcRequestValidator
+    MxAccessGrpcMapper
   Dashboard
     Pages
     Components
@@ -105,6 +105,15 @@ service MxAccessGateway {
 }
 ```
 
+`MxAccessGatewayService` implements these public RPCs in the gateway process.
+It validates public requests with `MxAccessGrpcRequestValidator`, delegates
+session lifecycle and command routing to `ISessionManager`, and maps worker
+command replies and events through `MxAccessGrpcMapper`. Session lookup,
+validation, and worker transport failures become gRPC status errors. MXAccess
+method replies that reached the worker remain `MxCommandReply` payloads so
+HRESULT values, status arrays, and method-specific reply fields survive
+transport boundaries.
+
 Add this later only after the command and event model is stable:
 
 ```protobuf
@@ -197,13 +206,23 @@ accounting and a clear fan-out policy.
 Behavior:
 
 1. Validate session id and authorize event access.
-2. Attach a stream cursor to the session event channel.
-3. Send events in worker sequence order.
-4. Stop on client cancellation, session close, or session fault.
-5. Emit a terminal status when the session faults if gRPC status alone cannot
+2. Attach the single active subscriber lease for the session.
+3. Read worker events into a bounded public stream queue.
+4. Send events in worker sequence order.
+5. Stop on client cancellation, session close, or session fault.
+6. Emit a terminal status when the session faults if gRPC status alone cannot
    preserve the required details.
 
-The gateway must not reorder events from one worker.
+`EventStreamService` owns subscriber tracking and public stream backpressure.
+The default policy allows one active subscriber per session. A second subscriber
+is rejected with `EventSubscriberAlreadyActive`. Stream cancellation releases
+the subscriber lease so a later stream can attach to the session.
+
+The gateway must not reorder events from one worker. `EventStreamService` writes
+mapped events to a bounded first-in, first-out queue and faults the session with
+`EventQueueOverflow` if the queue fills. The gateway does not synthesize
+`OperationComplete`; it forwards that family only when the worker reports a
+native MXAccess `OperationComplete` event.
 
 ## Web Dashboard
 
@@ -330,6 +349,20 @@ The worker remains authoritative for MXAccess handles. The gateway may keep a
 shadow state for diagnostics, but it must not invent, rewrite, or recycle
 MXAccess handles.
 
+`SessionManager` owns the current in-memory session registry. It allocates a
+session id, creates the worker pipe name and nonce, registers the session before
+worker startup, and removes the session if startup fails. A successful
+`OpenSession` attaches the ready `IWorkerClient` and transitions the session to
+`Ready`.
+
+Only `Ready` sessions accept command and event operations. `CloseSession` is
+idempotent for sessions still known to the registry: the first close shuts down
+the worker, and later closes return the final `Closed` state. Lease handling is
+exposed as a session hook so a monitor can close expired sessions without
+embedding lease policy in the worker client. Gateway shutdown walks the
+registry, closes each known session, and kills a worker if graceful shutdown
+fails.
+
 ## Worker Launch
 
 The gateway should launch the worker using explicit configuration:
@@ -561,7 +594,8 @@ worker MXAccess event
   -> worker outbound event queue
   -> worker pipe writer
   -> gateway read loop
-  -> session event channel
+  -> worker client event queue
+  -> EventStreamService bounded stream queue
   -> gRPC StreamEvents
 ```
 
@@ -575,13 +609,15 @@ The gateway should record:
 
 Default backpressure policy for parity testing should be fail-fast:
 
-1. If the session event channel fills, fault the session.
+1. If the worker client event queue fills, fault the worker client.
+2. If the public stream queue fills, fault the gateway session.
 2. Preserve the overflow details in logs and metrics.
 3. Do not silently drop data-change events.
 
-Do not set a production event-rate target before measurement. Emit event rate,
-queue depth, stream send latency, and overflow metrics. Later production modes
-may support explicit coalescing by item handle as an opt-in behavior.
+Do not set a production event-rate target before measurement. `GatewayMetrics`
+records received event counts by family, queue depth, stream disconnects, and
+overflow counts. Later production modes may support explicit coalescing by item
+handle as an opt-in behavior.
 
 The gateway should not synthesize `OperationComplete` from write completion,
 command replies, ASB completion queues, or completion-only status frames. Forward

docs/implementation-plan-mxaccess-worker.md

@@ -189,6 +189,8 @@ Tests:
 
 Labels: `area:worker`, `type:feature`, `priority:p0`
 
+Status: implemented.
+
 Deliverables:
 
 - `Register`,
@@ -447,4 +449,3 @@ Acceptance criteria:
 
 - each public method has planned parity fixture or documented gap,
 - gateway results preserve HRESULT/status/value/event shape.
-

docs/mxaccess-worker-instance-design.md

@@ -250,6 +250,17 @@ The loop should update a heartbeat timestamp after:
 - finishing a command,
 - processing an MXAccess event.
 
+`StaRuntime` implements this runtime boundary in the worker. It starts one
+background thread named `MxGateway.Worker.STA`, sets it to `ApartmentState.STA`,
+initializes COM through `StaComApartmentInitializer`, and runs
+`StaMessagePump`. Commands are scheduled through `InvokeAsync`; the command
+queue signals an `AutoResetEvent` so `MsgWaitForMultipleObjectsEx` can wake the
+STA without busy-waiting. `LastActivityUtc` records pump, command, startup, and
+shutdown activity so the future heartbeat/watchdog can report whether the STA
+is still responsive. Shutdown marks the runtime as closing, wakes the pump,
+rejects new commands, cancels queued work, uninitializes COM on the STA, and
+waits for the thread to exit.
+
 ## COM Creation
 
 The MXAccess analysis source at `C:\Users\dohertj2\Desktop\mxaccess` identifies
@@ -278,6 +289,16 @@ The worker should reference the interop assembly and instantiate
 `LMXProxyServerClass` on the dedicated STA thread. Keep the ProgID and assembly
 path configurable for diagnostics, but this COM class is the v1 default.
 
+`MxAccessStaSession` owns the initial COM creation path. It starts `StaRuntime`,
+creates `LMXProxyServerClass` through `MxAccessComObjectFactory` on the STA,
+attaches `MxAccessBaseEventSink`, and returns `WorkerReady` only after those
+steps succeed. `MxAccessSession` keeps the raw COM object private, records the
+STA managed thread id that created it, detaches the base event sink during
+disposal, and releases the COM reference on the STA. After creation,
+`MxAccessStaSession` owns a `StaCommandDispatcher` backed by
+`MxAccessCommandExecutor`; `DispatchAsync` queues contract commands back to the
+same STA instead of exposing the COM object to callers.
+
 Creation rules:
 
 - Create COM object only on the STA.
@@ -295,6 +316,11 @@ If COM creation fails, the worker should send a structured fault with:
 - worker process id,
 - session id.
 
+`WorkerPipeSession` maps startup exceptions from this path to
+`WorkerFaultCategory.MxaccessCreationFailed`, includes the captured HRESULT
+when the exception exposes one, and does not send `WorkerReady` after a failed
+COM creation attempt.
+
 ## Event Sink
 
 The worker must subscribe to every public MXAccess event family:
@@ -391,6 +417,21 @@ Diagnostics:
 Implement method-specific dispatch instead of a generic string method invoker.
 Parity tests need stable command-specific request and reply shapes.
 
+`MxAccessCommandExecutor` implements the first command pair:
+
+- `Register` calls `LMXProxyServerClass.Register` with the requested client
+  name and preserves the returned server handle in both `ReturnValue` and
+  `RegisterReply.ServerHandle`.
+- `Unregister` calls `LMXProxyServerClass.Unregister` with the requested server
+  handle. The reply has no method-specific payload because the public MXAccess
+  method returns `void`.
+
+Both commands set `Hresult` to `0` only after the COM call returns normally.
+COM exceptions flow through `StaCommandDispatcher`, which captures the thrown
+HRESULT and converts the reply to `ProtocolStatusCode.MxaccessFailure`.
+`MxAccessStaSession.GetRegisteredServerHandlesAsync` returns an STA-read
+snapshot of tracked server handles for diagnostics and future cleanup logic.
+
 ## Handle Registry
 
 The worker should track MXAccess state for diagnostics and cleanup, while still
@@ -411,6 +452,8 @@ Rules:
 
 - Do not invent handles.
 - Do not rewrite handles returned by MXAccess.
+- Record server handles only after `Register` succeeds.
+- Remove server handles only after `Unregister` succeeds.
 - Preserve invalid-handle behavior from MXAccess.
 - Preserve cross-server handle behavior from MXAccess.
 - Use registry state for cleanup and diagnostics, not semantic correction.

gateway.md

@@ -107,6 +107,8 @@ worker, correlation, command, and client identity fields with redaction applied
 before values enter log state. `GatewayMetrics` exposes counters, gauges, and
 histograms through .NET `Meter` and a snapshot API that dashboard services can
 project without binding to a metrics exporter.
+`DashboardSnapshotService` projects sessions, workers, metrics, faults, and
+effective configuration into immutable DTOs for read-only dashboard rendering.
 
 ### Worker Process
 
@@ -518,11 +520,7 @@ Worker policy:
 
 - bounded outbound event channel,
 - never block MXAccess event handler on pipe writes,
-- if the outbound channel is full, apply configured policy:
-  - disconnect session,
-  - drop oldest low-priority data-change events,
-  - coalesce data changes by item handle,
-  - or block briefly then fault.
+- fail the worker session when the outbound channel is full.
 
 For full parity testing, default should be fail-fast rather than silent drop.
 For production high-rate telemetry, add explicit coalescing modes.
@@ -531,9 +529,15 @@ Gateway policy:
 
 - one event sequencer per session,
 - preserve per-session event order,
-- support multiple client event subscribers only if explicitly required,
-- apply backpressure from slow gRPC streams,
-- disconnect or coalesce according to client-selected mode.
+- allow one active client event subscriber per session,
+- reject a second subscriber with a clear session error,
+- use a bounded `EventStreamService` queue between worker events and gRPC
+  writes,
+- fault the session when the bounded stream queue overflows,
+- detach the subscriber when the stream is canceled.
+
+The gateway forwards only events reported by the worker. It does not synthesize
+`OperationComplete` from write completion, command replies, or status frames.
 
 ## Isolation And Fault Handling
 
@@ -799,6 +803,12 @@ Core operations:
 - track worker state,
 - close or kill worker.
 
+The gateway implementation keeps sessions in an in-memory `SessionRegistry`
+keyed by session id. `SessionManager` owns the state machine, creates
+per-session pipe names and nonces, starts the worker through the worker-client
+factory, gates commands to `Ready` sessions, exposes lease-close hooks, and
+cleans up workers during gateway shutdown.
+
 State machine:
 
 ```text
@@ -846,6 +856,15 @@ The gRPC layer should be thin:
 Avoid embedding MXAccess-specific business logic in gRPC handlers. Keep the
 translation code testable.
 
+The gateway maps `MxAccessGateway` to `MxAccessGatewayService`. The service
+implements `OpenSession`, `CloseSession`, `Invoke`, and `StreamEvents` by
+validating public requests, delegating session work to `ISessionManager`, and
+using explicit mapper code for public-to-worker commands and worker replies.
+`StreamEvents` delegates subscriber ownership, ordering, and backpressure to
+`EventStreamService`. Missing sessions and transport failures return gRPC
+status errors; worker command replies preserve MXAccess HRESULT and status
+details in the public reply.
+
 ## C# Worker Versus C++ Worker
 
 Start with a C# .NET Framework 4.8 x86 worker.

src/MxGateway.Server/Dashboard/DashboardFaultSummary.cs

@@ -0,0 +1,9 @@
+namespace MxGateway.Server.Dashboard;
+
+public sealed record DashboardFaultSummary(
+    string Source,
+    string? SessionId,
+    int? WorkerProcessId,
+    string State,
+    string Message,
+    DateTimeOffset ObservedAt);

src/MxGateway.Server/Dashboard/DashboardMetricSummary.cs

@@ -0,0 +1,6 @@
+namespace MxGateway.Server.Dashboard;
+
+public sealed record DashboardMetricSummary(
+    string Name,
+    long Value,
+    string? Dimension = null);

src/MxGateway.Server/Dashboard/DashboardRedactor.cs

@@ -0,0 +1,34 @@
+using MxGateway.Server.Diagnostics;
+
+namespace MxGateway.Server.Dashboard;
+
+internal static class DashboardRedactor
+{
+    private static readonly string[] SensitiveTextMarkers =
+    [
+        "apikey",
+        "api_key",
+        "authorization",
+        "credential",
+        "password",
+        "secret",
+        "token",
+    ];
+
+    public static string? Redact(string? value)
+    {
+        if (string.IsNullOrWhiteSpace(value))
+        {
+            return value;
+        }
+
+        if (value.Contains("mxgw_", StringComparison.OrdinalIgnoreCase))
+        {
+            return GatewayLogRedactor.RedactClientIdentity(value);
+        }
+
+        return SensitiveTextMarkers.Any(marker => value.Contains(marker, StringComparison.OrdinalIgnoreCase))
+            ? GatewayLogRedactor.RedactedValue
+            : value;
+    }
+}

src/MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs

@@ -0,0 +1,11 @@
+namespace MxGateway.Server.Dashboard;
+
+public static class DashboardServiceCollectionExtensions
+{
+    public static IServiceCollection AddGatewayDashboard(this IServiceCollection services)
+    {
+        services.AddSingleton<IDashboardSnapshotService, DashboardSnapshotService>();
+
+        return services;
+    }
+}

src/MxGateway.Server/Dashboard/DashboardSessionSummary.cs

@@ -0,0 +1,19 @@
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Server.Dashboard;
+
+public sealed record DashboardSessionSummary(
+    string SessionId,
+    string BackendName,
+    SessionState State,
+    string? ClientIdentity,
+    string? ClientSessionName,
+    string? ClientCorrelationId,
+    DateTimeOffset OpenedAt,
+    DateTimeOffset LastClientActivityAt,
+    DateTimeOffset? LeaseExpiresAt,
+    int? WorkerProcessId,
+    WorkerClientState? WorkerState,
+    DateTimeOffset? LastWorkerHeartbeatAt,
+    string? LastFault);

src/MxGateway.Server/Dashboard/DashboardSnapshot.cs

@@ -0,0 +1,15 @@
+using MxGateway.Server.Configuration;
+
+namespace MxGateway.Server.Dashboard;
+
+public sealed record DashboardSnapshot(
+    DateTimeOffset GeneratedAt,
+    DateTimeOffset GatewayStartedAt,
+    TimeSpan GatewayUptime,
+    string GatewayStatus,
+    string GatewayVersion,
+    IReadOnlyList<DashboardSessionSummary> Sessions,
+    IReadOnlyList<DashboardWorkerSummary> Workers,
+    IReadOnlyList<DashboardMetricSummary> Metrics,
+    IReadOnlyList<DashboardFaultSummary> Faults,
+    EffectiveGatewayConfiguration Configuration);

src/MxGateway.Server/Dashboard/DashboardSnapshotService.cs

@@ -0,0 +1,196 @@
+using System.Runtime.CompilerServices;
+using Microsoft.Extensions.Options;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Metrics;
+using MxGateway.Server.Sessions;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Server.Dashboard;
+
+public sealed class DashboardSnapshotService : IDashboardSnapshotService
+{
+    private const string HealthyStatus = "Healthy";
+
+    private readonly ISessionRegistry _sessionRegistry;
+    private readonly GatewayMetrics _metrics;
+    private readonly IGatewayConfigurationProvider _configurationProvider;
+    private readonly TimeProvider _timeProvider;
+    private readonly DateTimeOffset _gatewayStartedAt;
+    private readonly TimeSpan _snapshotInterval;
+    private readonly int _recentFaultLimit;
+    private readonly int _recentSessionLimit;
+
+    public DashboardSnapshotService(
+        ISessionRegistry sessionRegistry,
+        GatewayMetrics metrics,
+        IGatewayConfigurationProvider configurationProvider,
+        IOptions<GatewayOptions> options,
+        TimeProvider? timeProvider = null)
+    {
+        _sessionRegistry = sessionRegistry ?? throw new ArgumentNullException(nameof(sessionRegistry));
+        _metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
+        _configurationProvider = configurationProvider ?? throw new ArgumentNullException(nameof(configurationProvider));
+        ArgumentNullException.ThrowIfNull(options);
+
+        _timeProvider = timeProvider ?? TimeProvider.System;
+        _gatewayStartedAt = _timeProvider.GetUtcNow();
+        _snapshotInterval = TimeSpan.FromMilliseconds(options.Value.Dashboard.SnapshotIntervalMilliseconds);
+        _recentFaultLimit = options.Value.Dashboard.RecentFaultLimit;
+        _recentSessionLimit = options.Value.Dashboard.RecentSessionLimit;
+    }
+
+    public DashboardSnapshot GetSnapshot()
+    {
+        DateTimeOffset generatedAt = _timeProvider.GetUtcNow();
+        IReadOnlyList<GatewaySession> sessions = _sessionRegistry.Snapshot()
+            .OrderByDescending(session => session.OpenedAt)
+            .ToArray();
+        IReadOnlyList<DashboardSessionSummary> sessionSummaries = sessions
+            .Take(ResolveLimit(_recentSessionLimit))
+            .Select(CreateSessionSummary)
+            .ToArray();
+        IReadOnlyList<DashboardWorkerSummary> workerSummaries = sessions
+            .Where(session => session.WorkerClient is not null)
+            .Select(CreateWorkerSummary)
+            .ToArray();
+        GatewayMetricsSnapshot metricsSnapshot = _metrics.GetSnapshot();
+
+        return new DashboardSnapshot(
+            GeneratedAt: generatedAt,
+            GatewayStartedAt: _gatewayStartedAt,
+            GatewayUptime: generatedAt - _gatewayStartedAt,
+            GatewayStatus: HealthyStatus,
+            GatewayVersion: typeof(DashboardSnapshotService).Assembly.GetName().Version?.ToString() ?? "unknown",
+            Sessions: sessionSummaries,
+            Workers: workerSummaries,
+            Metrics: CreateMetricSummaries(metricsSnapshot),
+            Faults: CreateFaultSummaries(sessions, generatedAt),
+            Configuration: _configurationProvider.GetEffectiveConfiguration());
+    }
+
+    public async IAsyncEnumerable<DashboardSnapshot> WatchSnapshotsAsync(
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        if (cancellationToken.IsCancellationRequested)
+        {
+            yield break;
+        }
+
+        yield return GetSnapshot();
+
+        using PeriodicTimer timer = new(_snapshotInterval, _timeProvider);
+        while (true)
+        {
+            bool hasNext;
+            try
+            {
+                hasNext = await timer.WaitForNextTickAsync(cancellationToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException)
+            {
+                yield break;
+            }
+
+            if (!hasNext)
+            {
+                yield break;
+            }
+
+            yield return GetSnapshot();
+        }
+    }
+
+    private static DashboardSessionSummary CreateSessionSummary(GatewaySession session)
+    {
+        IWorkerClient? workerClient = session.WorkerClient;
+
+        return new DashboardSessionSummary(
+            SessionId: session.SessionId,
+            BackendName: session.BackendName,
+            State: session.State,
+            ClientIdentity: DashboardRedactor.Redact(session.ClientIdentity),
+            ClientSessionName: DashboardRedactor.Redact(session.ClientSessionName),
+            ClientCorrelationId: DashboardRedactor.Redact(session.ClientCorrelationId),
+            OpenedAt: session.OpenedAt,
+            LastClientActivityAt: session.LastClientActivityAt,
+            LeaseExpiresAt: session.LeaseExpiresAt,
+            WorkerProcessId: workerClient?.ProcessId,
+            WorkerState: workerClient?.State,
+            LastWorkerHeartbeatAt: workerClient?.LastHeartbeatAt,
+            LastFault: DashboardRedactor.Redact(session.FinalFault));
+    }
+
+    private static DashboardWorkerSummary CreateWorkerSummary(GatewaySession session)
+    {
+        IWorkerClient workerClient = session.WorkerClient!;
+
+        return new DashboardWorkerSummary(
+            SessionId: session.SessionId,
+            ProcessId: workerClient.ProcessId,
+            State: workerClient.State,
+            LastHeartbeatAt: workerClient.LastHeartbeatAt,
+            LastFault: DashboardRedactor.Redact(session.FinalFault));
+    }
+
+    private static IReadOnlyList<DashboardMetricSummary> CreateMetricSummaries(GatewayMetricsSnapshot snapshot)
+    {
+        List<DashboardMetricSummary> metrics =
+        [
+            new("mxgateway.sessions.open", snapshot.OpenSessions),
+            new("mxgateway.workers.running", snapshot.WorkersRunning),
+            new("mxgateway.events.queue.depth", snapshot.EventQueueDepth),
+            new("mxgateway.sessions.opened", snapshot.SessionsOpened),
+            new("mxgateway.sessions.closed", snapshot.SessionsClosed),
+            new("mxgateway.commands.started", snapshot.CommandsStarted),
+            new("mxgateway.commands.succeeded", snapshot.CommandsSucceeded),
+            new("mxgateway.commands.failed", snapshot.CommandsFailed),
+            new("mxgateway.events.received", snapshot.EventsReceived),
+            new("mxgateway.queues.overflows", snapshot.QueueOverflows),
+            new("mxgateway.faults", snapshot.Faults),
+            new("mxgateway.workers.killed", snapshot.WorkerKills),
+            new("mxgateway.workers.exited", snapshot.WorkerExits),
+            new("mxgateway.heartbeats.failed", snapshot.HeartbeatFailures),
+            new("mxgateway.grpc.streams.disconnected", snapshot.StreamDisconnects),
+        ];
+
+        metrics.AddRange(snapshot.CommandFailuresByMethod
+            .OrderBy(entry => entry.Key, StringComparer.OrdinalIgnoreCase)
+            .Select(entry => new DashboardMetricSummary("mxgateway.commands.failed", entry.Value, entry.Key)));
+        metrics.AddRange(snapshot.EventsByFamily
+            .OrderBy(entry => entry.Key, StringComparer.OrdinalIgnoreCase)
+            .Select(entry => new DashboardMetricSummary("mxgateway.events.received", entry.Value, entry.Key)));
+
+        return metrics;
+    }
+
+    private IReadOnlyList<DashboardFaultSummary> CreateFaultSummaries(
+        IReadOnlyList<GatewaySession> sessions,
+        DateTimeOffset generatedAt)
+    {
+        return sessions
+            .Where(HasFault)
+            .Take(ResolveLimit(_recentFaultLimit))
+            .Select(session => new DashboardFaultSummary(
+                Source: session.WorkerClient?.State == WorkerClientState.Faulted ? "Worker" : "Session",
+                SessionId: session.SessionId,
+                WorkerProcessId: session.WorkerProcessId,
+                State: session.WorkerClient?.State == WorkerClientState.Faulted
+                    ? WorkerClientState.Faulted.ToString()
+                    : session.State.ToString(),
+                Message: DashboardRedactor.Redact(session.FinalFault) ?? "Faulted",
+                ObservedAt: generatedAt))
+            .ToArray();
+    }
+
+    private static bool HasFault(GatewaySession session)
+    {
+        return session.State == MxGateway.Contracts.Proto.SessionState.Faulted
+            || session.WorkerClient?.State == WorkerClientState.Faulted
+            || !string.IsNullOrWhiteSpace(session.FinalFault);
+    }
+
+    private static int ResolveLimit(int configuredLimit)
+    {
+        return configuredLimit < 0 ? 0 : configuredLimit;
+    }
+}

src/MxGateway.Server/Dashboard/DashboardWorkerSummary.cs

@@ -0,0 +1,10 @@
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Server.Dashboard;
+
+public sealed record DashboardWorkerSummary(
+    string SessionId,
+    int? ProcessId,
+    WorkerClientState State,
+    DateTimeOffset LastHeartbeatAt,
+    string? LastFault);

src/MxGateway.Server/Dashboard/IDashboardSnapshotService.cs

@@ -0,0 +1,8 @@
+namespace MxGateway.Server.Dashboard;
+
+public interface IDashboardSnapshotService
+{
+    DashboardSnapshot GetSnapshot();
+
+    IAsyncEnumerable<DashboardSnapshot> WatchSnapshotsAsync(CancellationToken cancellationToken);
+}

src/MxGateway.Server/GatewayApplication.cs

@@ -1,9 +1,12 @@
 using MxGateway.Contracts;
 using MxGateway.Server.Configuration;
+using MxGateway.Server.Dashboard;
 using MxGateway.Server.Diagnostics;
+using MxGateway.Server.Grpc;
 using MxGateway.Server.Metrics;
 using MxGateway.Server.Security.Authentication;
 using MxGateway.Server.Security.Authorization;
+using MxGateway.Server.Sessions;
 using MxGateway.Server.Workers;
 
 namespace MxGateway.Server;
@@ -30,7 +33,12 @@ public static class GatewayApplication
         builder.Services.AddGatewayGrpcAuthorization();
         builder.Services.AddHealthChecks();
         builder.Services.AddSingleton<GatewayMetrics>();
+        builder.Services.AddSingleton<MxAccessGrpcMapper>();
+        builder.Services.AddSingleton<MxAccessGrpcRequestValidator>();
+        builder.Services.AddSingleton<IEventStreamService, EventStreamService>();
         builder.Services.AddWorkerProcessLauncher();
+        builder.Services.AddGatewaySessions();
+        builder.Services.AddGatewayDashboard();
 
         return builder;
     }
@@ -47,6 +55,8 @@ public static class GatewayApplication
                     WorkerProtocolVersion: GatewayContractInfo.WorkerProtocolVersion)))
             .WithName("LiveHealth");
 
+        endpoints.MapGrpcService<MxAccessGatewayService>();
+
         return endpoints;
     }
 }

src/MxGateway.Server/Grpc/EventStreamService.cs

@@ -0,0 +1,140 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using Microsoft.Extensions.Options;
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Metrics;
+using MxGateway.Server.Sessions;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Server.Grpc;
+
+public sealed class EventStreamService(
+    ISessionManager sessionManager,
+    IOptions<GatewayOptions> options,
+    MxAccessGrpcMapper mapper,
+    GatewayMetrics metrics,
+    ILogger<EventStreamService> logger) : IEventStreamService
+{
+    public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
+        StreamEventsRequest request,
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        if (!sessionManager.TryGetSession(request.SessionId, out GatewaySession session))
+        {
+            throw new SessionManagerException(
+                SessionManagerErrorCode.SessionNotFound,
+                $"Session {request.SessionId} was not found.");
+        }
+
+        using IDisposable subscriber = session.AttachEventSubscriber(
+            options.Value.Sessions.AllowMultipleEventSubscribers);
+        using CancellationTokenSource streamCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+
+        int streamQueueDepth = 0;
+        Channel<MxEvent> eventQueue = Channel.CreateBounded<MxEvent>(
+            new BoundedChannelOptions(options.Value.Events.QueueCapacity)
+            {
+                SingleReader = true,
+                SingleWriter = true,
+                FullMode = BoundedChannelFullMode.Wait,
+                AllowSynchronousContinuations = false,
+            });
+        Task producerTask = ProduceEventsAsync(
+            session,
+            request.AfterWorkerSequence,
+            eventQueue.Writer,
+            () =>
+            {
+                int depth = Interlocked.Increment(ref streamQueueDepth);
+                metrics.SetEventQueueDepth(depth);
+            },
+            streamCts.Token);
+
+        try
+        {
+            await foreach (MxEvent mxEvent in eventQueue.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+            {
+                int depth = Math.Max(0, Interlocked.Decrement(ref streamQueueDepth));
+                metrics.SetEventQueueDepth(depth);
+                yield return mxEvent;
+            }
+
+            await producerTask.ConfigureAwait(false);
+        }
+        finally
+        {
+            await streamCts.CancelAsync().ConfigureAwait(false);
+            subscriber.Dispose();
+            metrics.StreamDisconnected("Detached");
+
+            try
+            {
+                await producerTask.ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (streamCts.IsCancellationRequested)
+            {
+            }
+            catch (Exception exception)
+            {
+                logger.LogDebug(
+                    exception,
+                    "Event stream producer stopped for session {SessionId}.",
+                    request.SessionId);
+            }
+        }
+    }
+
+    private async Task ProduceEventsAsync(
+        GatewaySession session,
+        ulong afterWorkerSequence,
+        ChannelWriter<MxEvent> writer,
+        Action eventQueued,
+        CancellationToken cancellationToken)
+    {
+        try
+        {
+            await foreach (WorkerEvent workerEvent in session
+                .ReadEventsAsync(cancellationToken)
+                .WithCancellation(cancellationToken)
+                .ConfigureAwait(false))
+            {
+                MxEvent publicEvent = mapper.MapEvent(workerEvent);
+                if (publicEvent.WorkerSequence <= afterWorkerSequence)
+                {
+                    continue;
+                }
+
+                if (!writer.TryWrite(publicEvent))
+                {
+                    string message = $"Session {session.SessionId} event stream queue overflowed.";
+                    session.MarkFaulted(message);
+                    metrics.QueueOverflow("grpc-event-stream");
+                    metrics.Fault(SessionManagerErrorCode.EventQueueOverflow.ToString());
+                    writer.TryComplete(new SessionManagerException(
+                        SessionManagerErrorCode.EventQueueOverflow,
+                        message));
+                    return;
+                }
+
+                eventQueued();
+            }
+
+            writer.TryComplete();
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            writer.TryComplete();
+        }
+        catch (Exception exception)
+        {
+            if (exception is WorkerClientException)
+            {
+                session.MarkFaulted(exception.Message);
+                metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString());
+            }
+
+            writer.TryComplete(exception);
+        }
+    }
+}

src/MxGateway.Server/Grpc/IEventStreamService.cs

@@ -0,0 +1,10 @@
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Server.Grpc;
+
+public interface IEventStreamService
+{
+    IAsyncEnumerable<MxEvent> StreamEventsAsync(
+        StreamEventsRequest request,
+        CancellationToken cancellationToken);
+}

src/MxGateway.Server/Grpc/MxAccessGatewayService.cs

@@ -0,0 +1,176 @@
+using Grpc.Core;
+using MxGateway.Contracts;
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Security.Authorization;
+using MxGateway.Server.Sessions;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Server.Grpc;
+
+public sealed class MxAccessGatewayService(
+    ISessionManager sessionManager,
+    IGatewayRequestIdentityAccessor identityAccessor,
+    MxAccessGrpcRequestValidator requestValidator,
+    MxAccessGrpcMapper mapper,
+    IEventStreamService eventStreamService,
+    ILogger<MxAccessGatewayService> logger) : MxAccessGateway.MxAccessGatewayBase
+{
+    public override async Task<OpenSessionReply> OpenSession(
+        OpenSessionRequest request,
+        ServerCallContext context)
+    {
+        try
+        {
+            requestValidator.ValidateOpenSession(request);
+            GatewaySession session = await sessionManager
+                .OpenSessionAsync(
+                    SessionOpenRequest.FromContract(request),
+                    ResolveClientIdentity(),
+                    context.CancellationToken)
+                .ConfigureAwait(false);
+
+            OpenSessionReply reply = new()
+            {
+                SessionId = session.SessionId,
+                BackendName = session.BackendName,
+                WorkerProcessId = session.WorkerProcessId ?? 0,
+                WorkerProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
+                DefaultCommandTimeout = Google.Protobuf.WellKnownTypes.Duration.FromTimeSpan(session.CommandTimeout),
+                ProtocolStatus = MxAccessGrpcMapper.Ok(),
+            };
+            reply.Capabilities.Add("unary-open-session");
+            reply.Capabilities.Add("unary-close-session");
+            reply.Capabilities.Add("unary-invoke");
+            reply.Capabilities.Add("server-stream-events");
+
+            return reply;
+        }
+        catch (Exception exception) when (exception is not RpcException)
+        {
+            throw MapException(exception);
+        }
+    }
+
+    public override async Task<CloseSessionReply> CloseSession(
+        CloseSessionRequest request,
+        ServerCallContext context)
+    {
+        try
+        {
+            requestValidator.ValidateCloseSession(request);
+            SessionCloseResult result = await sessionManager
+                .CloseSessionAsync(request.SessionId, context.CancellationToken)
+                .ConfigureAwait(false);
+
+            return new CloseSessionReply
+            {
+                SessionId = result.SessionId,
+                FinalState = result.FinalState,
+                ProtocolStatus = MxAccessGrpcMapper.Ok(result.AlreadyClosed ? "Session was already closed." : "Session closed."),
+            };
+        }
+        catch (Exception exception) when (exception is not RpcException)
+        {
+            throw MapException(exception);
+        }
+    }
+
+    public override async Task<MxCommandReply> Invoke(
+        MxCommandRequest request,
+        ServerCallContext context)
+    {
+        try
+        {
+            requestValidator.ValidateInvoke(request);
+            WorkerCommand workerCommand = mapper.MapCommand(request);
+            WorkerCommandReply workerReply = await sessionManager
+                .InvokeAsync(request.SessionId, workerCommand, context.CancellationToken)
+                .ConfigureAwait(false);
+
+            return mapper.MapCommandReply(workerReply);
+        }
+        catch (Exception exception) when (exception is not RpcException)
+        {
+            throw MapException(exception);
+        }
+    }
+
+    public override async Task StreamEvents(
+        StreamEventsRequest request,
+        IServerStreamWriter<MxEvent> responseStream,
+        ServerCallContext context)
+    {
+        try
+        {
+            requestValidator.ValidateStreamEvents(request);
+            await foreach (MxEvent publicEvent in eventStreamService
+                .StreamEventsAsync(request, context.CancellationToken)
+                .WithCancellation(context.CancellationToken)
+                .ConfigureAwait(false))
+            {
+                await responseStream.WriteAsync(publicEvent).ConfigureAwait(false);
+            }
+        }
+        catch (Exception exception) when (exception is not RpcException)
+        {
+            throw MapException(exception);
+        }
+    }
+
+    private string? ResolveClientIdentity()
+    {
+        return identityAccessor.Current?.DisplayName ?? identityAccessor.Current?.KeyId;
+    }
+
+    private RpcException MapException(Exception exception)
+    {
+        if (exception is OperationCanceledException)
+        {
+            return new RpcException(new Status(StatusCode.Cancelled, "gRPC request was canceled."));
+        }
+
+        if (exception is SessionManagerException sessionException)
+        {
+            return MapSessionException(sessionException);
+        }
+
+        if (exception is WorkerClientException workerClientException)
+        {
+            return MapWorkerClientException(workerClientException);
+        }
+
+        logger.LogWarning(exception, "Public gRPC request failed.");
+        return new RpcException(new Status(StatusCode.Unavailable, "Gateway request failed before an MXAccess reply was available."));
+    }
+
+    private static RpcException MapSessionException(SessionManagerException exception)
+    {
+        StatusCode statusCode = exception.ErrorCode switch
+        {
+            SessionManagerErrorCode.SessionNotFound => StatusCode.NotFound,
+            SessionManagerErrorCode.SessionNotReady => StatusCode.FailedPrecondition,
+            SessionManagerErrorCode.EventSubscriberAlreadyActive => StatusCode.ResourceExhausted,
+            SessionManagerErrorCode.EventQueueOverflow => StatusCode.ResourceExhausted,
+            SessionManagerErrorCode.SessionLimitExceeded => StatusCode.ResourceExhausted,
+            SessionManagerErrorCode.OpenFailed => StatusCode.Unavailable,
+            SessionManagerErrorCode.CloseFailed => StatusCode.Unavailable,
+            _ => StatusCode.Unavailable,
+        };
+
+        return new RpcException(new Status(statusCode, exception.Message));
+    }
+
+    private static RpcException MapWorkerClientException(WorkerClientException exception)
+    {
+        StatusCode statusCode = exception.ErrorCode switch
+        {
+            WorkerClientErrorCode.CommandTimeout => StatusCode.DeadlineExceeded,
+            WorkerClientErrorCode.GatewayShutdown => StatusCode.Cancelled,
+            WorkerClientErrorCode.InvalidState => StatusCode.FailedPrecondition,
+            WorkerClientErrorCode.ProtocolViolation => StatusCode.Internal,
+            _ => StatusCode.Unavailable,
+        };
+
+        return new RpcException(new Status(statusCode, exception.Message));
+    }
+}

src/MxGateway.Server/Grpc/MxAccessGrpcMapper.cs

@@ -0,0 +1,124 @@
+using Google.Protobuf.WellKnownTypes;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Server.Grpc;
+
+public sealed class MxAccessGrpcMapper
+{
+    private readonly TimeProvider _timeProvider;
+
+    public MxAccessGrpcMapper(TimeProvider? timeProvider = null)
+    {
+        _timeProvider = timeProvider ?? TimeProvider.System;
+    }
+
+    public WorkerCommand MapCommand(MxCommandRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+        ArgumentNullException.ThrowIfNull(request.Command);
+
+        return new WorkerCommand
+        {
+            Command = request.Command.Clone(),
+            EnqueueTimestamp = Timestamp.FromDateTimeOffset(_timeProvider.GetUtcNow()),
+        };
+    }
+
+    public MxCommandReply MapCommandReply(WorkerCommandReply reply)
+    {
+        ArgumentNullException.ThrowIfNull(reply);
+
+        if (reply.Reply is null)
+        {
+            return new MxCommandReply
+            {
+                ProtocolStatus = ProtocolViolation("Worker command reply did not contain a public reply payload."),
+            };
+        }
+
+        return reply.Reply.Clone();
+    }
+
+    public MxEvent MapEvent(WorkerEvent workerEvent)
+    {
+        ArgumentNullException.ThrowIfNull(workerEvent);
+
+        return workerEvent.Event?.Clone() ?? new MxEvent
+        {
+            Family = MxEventFamily.Unspecified,
+            RawStatus = "Worker event did not contain a public event payload.",
+        };
+    }
+
+    public static ProtocolStatus Ok(string message = "OK")
+    {
+        return new ProtocolStatus
+        {
+            Code = ProtocolStatusCode.Ok,
+            Message = message,
+        };
+    }
+
+    public static ProtocolStatus InvalidRequest(string message)
+    {
+        return new ProtocolStatus
+        {
+            Code = ProtocolStatusCode.InvalidRequest,
+            Message = message,
+        };
+    }
+
+    public static ProtocolStatus SessionNotFound(string message)
+    {
+        return new ProtocolStatus
+        {
+            Code = ProtocolStatusCode.SessionNotFound,
+            Message = message,
+        };
+    }
+
+    public static ProtocolStatus SessionNotReady(string message)
+    {
+        return new ProtocolStatus
+        {
+            Code = ProtocolStatusCode.SessionNotReady,
+            Message = message,
+        };
+    }
+
+    public static ProtocolStatus WorkerUnavailable(string message)
+    {
+        return new ProtocolStatus
+        {
+            Code = ProtocolStatusCode.WorkerUnavailable,
+            Message = message,
+        };
+    }
+
+    public static ProtocolStatus Timeout(string message)
+    {
+        return new ProtocolStatus
+        {
+            Code = ProtocolStatusCode.Timeout,
+            Message = message,
+        };
+    }
+
+    public static ProtocolStatus Canceled(string message)
+    {
+        return new ProtocolStatus
+        {
+            Code = ProtocolStatusCode.Canceled,
+            Message = message,
+        };
+    }
+
+    public static ProtocolStatus ProtocolViolation(string message)
+    {
+        return new ProtocolStatus
+        {
+            Code = ProtocolStatusCode.ProtocolViolation,
+            Message = message,
+        };
+    }
+}

src/MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs

@@ -0,0 +1,101 @@
+using Grpc.Core;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Server.Grpc;
+
+public sealed class MxAccessGrpcRequestValidator
+{
+    public void ValidateOpenSession(OpenSessionRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+
+        if (request.CommandTimeout is not null && request.CommandTimeout.ToTimeSpan() <= TimeSpan.Zero)
+        {
+            throw InvalidArgument("Command timeout must be greater than zero when provided.");
+        }
+    }
+
+    public void ValidateCloseSession(CloseSessionRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+        RequireSessionId(request.SessionId);
+    }
+
+    public void ValidateStreamEvents(StreamEventsRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+        RequireSessionId(request.SessionId);
+    }
+
+    public void ValidateInvoke(MxCommandRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+        RequireSessionId(request.SessionId);
+
+        if (request.Command is null)
+        {
+            throw InvalidArgument("Invoke requires a command payload.");
+        }
+
+        if (request.Command.Kind is MxCommandKind.Unspecified)
+        {
+            throw InvalidArgument("Invoke requires a command kind.");
+        }
+
+        ValidateCommandPayload(request.Command);
+    }
+
+    private static void RequireSessionId(string sessionId)
+    {
+        if (string.IsNullOrWhiteSpace(sessionId))
+        {
+            throw InvalidArgument("Session id is required.");
+        }
+    }
+
+    private static void ValidateCommandPayload(MxCommand command)
+    {
+        MxCommand.PayloadOneofCase expectedPayload = ExpectedPayload(command.Kind);
+        if (command.PayloadCase != expectedPayload)
+        {
+            throw InvalidArgument(
+                $"Command kind {command.Kind} requires payload {expectedPayload} but received {command.PayloadCase}.");
+        }
+    }
+
+    private static MxCommand.PayloadOneofCase ExpectedPayload(MxCommandKind kind)
+    {
+        return kind switch
+        {
+            MxCommandKind.Register => MxCommand.PayloadOneofCase.Register,
+            MxCommandKind.Unregister => MxCommand.PayloadOneofCase.Unregister,
+            MxCommandKind.AddItem => MxCommand.PayloadOneofCase.AddItem,
+            MxCommandKind.AddItem2 => MxCommand.PayloadOneofCase.AddItem2,
+            MxCommandKind.RemoveItem => MxCommand.PayloadOneofCase.RemoveItem,
+            MxCommandKind.Advise => MxCommand.PayloadOneofCase.Advise,
+            MxCommandKind.UnAdvise => MxCommand.PayloadOneofCase.UnAdvise,
+            MxCommandKind.AdviseSupervisory => MxCommand.PayloadOneofCase.AdviseSupervisory,
+            MxCommandKind.AddBufferedItem => MxCommand.PayloadOneofCase.AddBufferedItem,
+            MxCommandKind.SetBufferedUpdateInterval => MxCommand.PayloadOneofCase.SetBufferedUpdateInterval,
+            MxCommandKind.Suspend => MxCommand.PayloadOneofCase.Suspend,
+            MxCommandKind.Activate => MxCommand.PayloadOneofCase.Activate,
+            MxCommandKind.Write => MxCommand.PayloadOneofCase.Write,
+            MxCommandKind.Write2 => MxCommand.PayloadOneofCase.Write2,
+            MxCommandKind.WriteSecured => MxCommand.PayloadOneofCase.WriteSecured,
+            MxCommandKind.WriteSecured2 => MxCommand.PayloadOneofCase.WriteSecured2,
+            MxCommandKind.AuthenticateUser => MxCommand.PayloadOneofCase.AuthenticateUser,
+            MxCommandKind.ArchestraUserToId => MxCommand.PayloadOneofCase.ArchestraUserToId,
+            MxCommandKind.Ping => MxCommand.PayloadOneofCase.Ping,
+            MxCommandKind.GetSessionState => MxCommand.PayloadOneofCase.GetSessionState,
+            MxCommandKind.GetWorkerInfo => MxCommand.PayloadOneofCase.GetWorkerInfo,
+            MxCommandKind.DrainEvents => MxCommand.PayloadOneofCase.DrainEvents,
+            MxCommandKind.ShutdownWorker => MxCommand.PayloadOneofCase.ShutdownWorker,
+            _ => MxCommand.PayloadOneofCase.None,
+        };
+    }
+
+    private static RpcException InvalidArgument(string detail)
+    {
+        return new RpcException(new Status(StatusCode.InvalidArgument, detail));
+    }
+}

src/MxGateway.Server/Sessions/GatewaySession.cs

@@ -0,0 +1,352 @@
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Server.Sessions;
+
+public sealed class GatewaySession
+{
+    private readonly object _syncRoot = new();
+    private readonly SemaphoreSlim _closeLock = new(1, 1);
+    private IWorkerClient? _workerClient;
+    private SessionState _state = SessionState.Creating;
+    private string? _finalFault;
+    private DateTimeOffset _lastClientActivityAt;
+    private DateTimeOffset? _leaseExpiresAt;
+    private bool _closeStarted;
+    private int _activeEventSubscriberCount;
+
+    public GatewaySession(
+        string sessionId,
+        string backendName,
+        string pipeName,
+        string nonce,
+        string? clientIdentity,
+        string? clientSessionName,
+        string? clientCorrelationId,
+        TimeSpan commandTimeout,
+        TimeSpan startupTimeout,
+        TimeSpan shutdownTimeout,
+        DateTimeOffset openedAt)
+    {
+        if (string.IsNullOrWhiteSpace(sessionId))
+        {
+            throw new ArgumentException("Session id is required.", nameof(sessionId));
+        }
+
+        if (string.IsNullOrWhiteSpace(backendName))
+        {
+            throw new ArgumentException("Backend name is required.", nameof(backendName));
+        }
+
+        if (string.IsNullOrWhiteSpace(pipeName))
+        {
+            throw new ArgumentException("Pipe name is required.", nameof(pipeName));
+        }
+
+        if (string.IsNullOrWhiteSpace(nonce))
+        {
+            throw new ArgumentException("Nonce is required.", nameof(nonce));
+        }
+
+        SessionId = sessionId;
+        BackendName = backendName;
+        PipeName = pipeName;
+        Nonce = nonce;
+        ClientIdentity = clientIdentity;
+        ClientSessionName = clientSessionName;
+        ClientCorrelationId = clientCorrelationId;
+        CommandTimeout = commandTimeout;
+        StartupTimeout = startupTimeout;
+        ShutdownTimeout = shutdownTimeout;
+        OpenedAt = openedAt;
+        _lastClientActivityAt = openedAt;
+    }
+
+    public string SessionId { get; }
+
+    public string BackendName { get; }
+
+    public string PipeName { get; }
+
+    public string Nonce { get; }
+
+    public string? ClientIdentity { get; }
+
+    public string? ClientSessionName { get; }
+
+    public string? ClientCorrelationId { get; }
+
+    public TimeSpan CommandTimeout { get; }
+
+    public TimeSpan StartupTimeout { get; }
+
+    public TimeSpan ShutdownTimeout { get; }
+
+    public DateTimeOffset OpenedAt { get; }
+
+    public int? WorkerProcessId => _workerClient?.ProcessId;
+
+    public IWorkerClient? WorkerClient => _workerClient;
+
+    public SessionState State
+    {
+        get
+        {
+            lock (_syncRoot)
+            {
+                return _state;
+            }
+        }
+    }
+
+    public DateTimeOffset LastClientActivityAt
+    {
+        get
+        {
+            lock (_syncRoot)
+            {
+                return _lastClientActivityAt;
+            }
+        }
+    }
+
+    public DateTimeOffset? LeaseExpiresAt
+    {
+        get
+        {
+            lock (_syncRoot)
+            {
+                return _leaseExpiresAt;
+            }
+        }
+    }
+
+    public string? FinalFault
+    {
+        get
+        {
+            lock (_syncRoot)
+            {
+                return _finalFault;
+            }
+        }
+    }
+
+    public int ActiveEventSubscriberCount
+    {
+        get
+        {
+            lock (_syncRoot)
+            {
+                return _activeEventSubscriberCount;
+            }
+        }
+    }
+
+    public void AttachWorkerClient(IWorkerClient workerClient)
+    {
+        ArgumentNullException.ThrowIfNull(workerClient);
+
+        lock (_syncRoot)
+        {
+            _workerClient = workerClient;
+        }
+    }
+
+    public void TransitionTo(SessionState nextState)
+    {
+        lock (_syncRoot)
+        {
+            if (_state is SessionState.Closed)
+            {
+                return;
+            }
+
+            if (_state is SessionState.Faulted && nextState is not SessionState.Closed)
+            {
+                return;
+            }
+
+            _state = nextState;
+        }
+    }
+
+    public void MarkReady()
+    {
+        TransitionTo(SessionState.Ready);
+    }
+
+    public void MarkFaulted(string reason)
+    {
+        lock (_syncRoot)
+        {
+            if (_state is SessionState.Closed)
+            {
+                return;
+            }
+
+            _finalFault = reason;
+            _state = SessionState.Faulted;
+        }
+    }
+
+    public void TouchClientActivity(DateTimeOffset activityAt)
+    {
+        lock (_syncRoot)
+        {
+            _lastClientActivityAt = activityAt;
+        }
+    }
+
+    public void ExtendLease(DateTimeOffset leaseExpiresAt)
+    {
+        lock (_syncRoot)
+        {
+            _leaseExpiresAt = leaseExpiresAt;
+        }
+    }
+
+    public bool IsLeaseExpired(DateTimeOffset now)
+    {
+        lock (_syncRoot)
+        {
+            return _leaseExpiresAt is not null && _leaseExpiresAt <= now;
+        }
+    }
+
+    public IDisposable AttachEventSubscriber(bool allowMultipleSubscribers)
+    {
+        lock (_syncRoot)
+        {
+            if (_state != SessionState.Ready || _workerClient?.State != WorkerClientState.Ready)
+            {
+                throw new SessionManagerException(
+                    SessionManagerErrorCode.SessionNotReady,
+                    $"Session {SessionId} is not ready for event streaming. Current state is {_state}.");
+            }
+
+            if (!allowMultipleSubscribers && _activeEventSubscriberCount > 0)
+            {
+                throw new SessionManagerException(
+                    SessionManagerErrorCode.EventSubscriberAlreadyActive,
+                    $"Session {SessionId} already has an active event stream subscriber.");
+            }
+
+            _activeEventSubscriberCount++;
+            return new EventSubscriberLease(this);
+        }
+    }
+
+    public async Task<WorkerCommandReply> InvokeAsync(
+        WorkerCommand command,
+        CancellationToken cancellationToken)
+    {
+        IWorkerClient workerClient = GetReadyWorkerClient();
+        TouchClientActivity(DateTimeOffset.UtcNow);
+
+        return await workerClient.InvokeAsync(command, CommandTimeout, cancellationToken).ConfigureAwait(false);
+    }
+
+    public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(CancellationToken cancellationToken)
+    {
+        IWorkerClient workerClient = GetReadyWorkerClient();
+        TouchClientActivity(DateTimeOffset.UtcNow);
+
+        return workerClient.ReadEventsAsync(cancellationToken);
+    }
+
+    public async Task<SessionCloseResult> CloseAsync(
+        string reason,
+        CancellationToken cancellationToken)
+    {
+        await _closeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            if (_state is SessionState.Closed)
+            {
+                return new SessionCloseResult(SessionId, SessionState.Closed, AlreadyClosed: true);
+            }
+
+            bool alreadyClosing = _closeStarted;
+            _closeStarted = true;
+            _state = SessionState.Closing;
+
+            if (_workerClient is not null)
+            {
+                try
+                {
+                    await _workerClient.ShutdownAsync(ShutdownTimeout, cancellationToken).ConfigureAwait(false);
+                }
+                catch
+                {
+                    _workerClient.Kill(reason);
+                    throw;
+                }
+            }
+
+            _state = SessionState.Closed;
+            return new SessionCloseResult(SessionId, SessionState.Closed, alreadyClosing);
+        }
+        finally
+        {
+            _closeLock.Release();
+        }
+    }
+
+    public void KillWorker(string reason)
+    {
+        _workerClient?.Kill(reason);
+        TransitionTo(SessionState.Closed);
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        _closeLock.Dispose();
+        if (_workerClient is not null)
+        {
+            await _workerClient.DisposeAsync().ConfigureAwait(false);
+        }
+    }
+
+    private IWorkerClient GetReadyWorkerClient()
+    {
+        lock (_syncRoot)
+        {
+            if (_state != SessionState.Ready || _workerClient?.State != WorkerClientState.Ready)
+            {
+                throw new SessionManagerException(
+                    SessionManagerErrorCode.SessionNotReady,
+                    $"Session {SessionId} is not ready. Current state is {_state}.");
+            }
+
+            return _workerClient;
+        }
+    }
+
+    private void DetachEventSubscriber()
+    {
+        lock (_syncRoot)
+        {
+            if (_activeEventSubscriberCount > 0)
+            {
+                _activeEventSubscriberCount--;
+            }
+        }
+    }
+
+    private sealed class EventSubscriberLease(GatewaySession session) : IDisposable
+    {
+        private bool _disposed;
+
+        public void Dispose()
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            session.DetachEventSubscriber();
+            _disposed = true;
+        }
+    }
+}

src/MxGateway.Server/Sessions/ISessionManager.cs

@@ -0,0 +1,34 @@
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Server.Sessions;
+
+public interface ISessionManager
+{
+    Task<GatewaySession> OpenSessionAsync(
+        SessionOpenRequest request,
+        string? clientIdentity,
+        CancellationToken cancellationToken);
+
+    bool TryGetSession(
+        string sessionId,
+        out GatewaySession session);
+
+    Task<WorkerCommandReply> InvokeAsync(
+        string sessionId,
+        WorkerCommand command,
+        CancellationToken cancellationToken);
+
+    IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+        string sessionId,
+        CancellationToken cancellationToken);
+
+    Task<SessionCloseResult> CloseSessionAsync(
+        string sessionId,
+        CancellationToken cancellationToken);
+
+    Task<int> CloseExpiredLeasesAsync(
+        DateTimeOffset now,
+        CancellationToken cancellationToken);
+
+    Task ShutdownAsync(CancellationToken cancellationToken);
+}

src/MxGateway.Server/Sessions/ISessionRegistry.cs

@@ -0,0 +1,16 @@
+namespace MxGateway.Server.Sessions;
+
+public interface ISessionRegistry
+{
+    int Count { get; }
+
+    int ActiveCount { get; }
+
+    bool TryAdd(GatewaySession session);
+
+    bool TryGet(string sessionId, out GatewaySession session);
+
+    bool TryRemove(string sessionId, out GatewaySession session);
+
+    IReadOnlyCollection<GatewaySession> Snapshot();
+}

src/MxGateway.Server/Sessions/ISessionWorkerClientFactory.cs

@@ -0,0 +1,8 @@
+namespace MxGateway.Server.Sessions;
+
+public interface ISessionWorkerClientFactory
+{
+    Task<MxGateway.Server.Workers.IWorkerClient> CreateAsync(
+        GatewaySession session,
+        CancellationToken cancellationToken);
+}

src/MxGateway.Server/Sessions/SessionCloseResult.cs

@@ -0,0 +1,8 @@
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Server.Sessions;
+
+public sealed record SessionCloseResult(
+    string SessionId,
+    SessionState FinalState,
+    bool AlreadyClosed);

src/MxGateway.Server/Sessions/SessionManager.cs

@@ -0,0 +1,287 @@
+using System.Security.Cryptography;
+using Google.Protobuf.WellKnownTypes;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using MxGateway.Contracts;
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Metrics;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Server.Sessions;
+
+public sealed class SessionManager : ISessionManager
+{
+    public const string DefaultCloseReason = "client-close";
+    public const string GatewayShutdownReason = "gateway-shutdown";
+    public const string LeaseExpiredReason = "lease-expired";
+
+    private readonly ISessionRegistry _registry;
+    private readonly ISessionWorkerClientFactory _workerClientFactory;
+    private readonly GatewayMetrics _metrics;
+    private readonly TimeProvider _timeProvider;
+    private readonly ILogger<SessionManager> _logger;
+    private readonly GatewayOptions _options;
+
+    public SessionManager(
+        ISessionRegistry registry,
+        ISessionWorkerClientFactory workerClientFactory,
+        IOptions<GatewayOptions> options,
+        GatewayMetrics metrics,
+        TimeProvider? timeProvider = null,
+        ILogger<SessionManager>? logger = null)
+    {
+        _registry = registry ?? throw new ArgumentNullException(nameof(registry));
+        _workerClientFactory = workerClientFactory ?? throw new ArgumentNullException(nameof(workerClientFactory));
+        ArgumentNullException.ThrowIfNull(options);
+        _metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
+        _timeProvider = timeProvider ?? TimeProvider.System;
+        _logger = logger ?? NullLogger<SessionManager>.Instance;
+        _options = options.Value;
+    }
+
+    public async Task<GatewaySession> OpenSessionAsync(
+        SessionOpenRequest request,
+        string? clientIdentity,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+        EnsureSessionCapacity();
+
+        GatewaySession session = CreateSession(request, clientIdentity);
+        if (!_registry.TryAdd(session))
+        {
+            throw new SessionManagerException(
+                SessionManagerErrorCode.OpenFailed,
+                $"Session id collision while opening session {session.SessionId}.");
+        }
+
+        try
+        {
+            session.TransitionTo(SessionState.StartingWorker);
+            IWorkerClient workerClient = await _workerClientFactory
+                .CreateAsync(session, cancellationToken)
+                .ConfigureAwait(false);
+
+            session.AttachWorkerClient(workerClient);
+            session.MarkReady();
+            _metrics.SessionOpened();
+
+            return session;
+        }
+        catch (Exception exception)
+        {
+            session.MarkFaulted(exception.Message);
+            _registry.TryRemove(session.SessionId, out _);
+            await session.DisposeAsync().ConfigureAwait(false);
+            _metrics.Fault(SessionManagerErrorCode.OpenFailed.ToString());
+            _logger.LogWarning(
+                exception,
+                "Failed to open gateway session {SessionId}.",
+                session.SessionId);
+
+            throw new SessionManagerException(
+                SessionManagerErrorCode.OpenFailed,
+                $"Failed to open session {session.SessionId}.",
+                exception);
+        }
+    }
+
+    public bool TryGetSession(
+        string sessionId,
+        out GatewaySession session)
+    {
+        return _registry.TryGet(sessionId, out session);
+    }
+
+    public async Task<WorkerCommandReply> InvokeAsync(
+        string sessionId,
+        WorkerCommand command,
+        CancellationToken cancellationToken)
+    {
+        GatewaySession session = GetRequiredSession(sessionId);
+
+        try
+        {
+            return await session.InvokeAsync(command, cancellationToken).ConfigureAwait(false);
+        }
+        catch (SessionManagerException)
+        {
+            throw;
+        }
+        catch (Exception exception)
+        {
+            if (session.WorkerClient?.State == WorkerClientState.Faulted)
+            {
+                session.MarkFaulted(exception.Message);
+            }
+
+            throw;
+        }
+    }
+
+    public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+        string sessionId,
+        CancellationToken cancellationToken)
+    {
+        GatewaySession session = GetRequiredSession(sessionId);
+
+        return session.ReadEventsAsync(cancellationToken);
+    }
+
+    public async Task<SessionCloseResult> CloseSessionAsync(
+        string sessionId,
+        CancellationToken cancellationToken)
+    {
+        GatewaySession session = GetRequiredSession(sessionId);
+        SessionCloseResult result = await CloseSessionCoreAsync(
+            session,
+            DefaultCloseReason,
+            cancellationToken).ConfigureAwait(false);
+
+        return result;
+    }
+
+    public async Task<int> CloseExpiredLeasesAsync(
+        DateTimeOffset now,
+        CancellationToken cancellationToken)
+    {
+        int closedCount = 0;
+        foreach (GatewaySession session in _registry.Snapshot())
+        {
+            if (!session.IsLeaseExpired(now))
+            {
+                continue;
+            }
+
+            await CloseSessionCoreAsync(session, LeaseExpiredReason, cancellationToken).ConfigureAwait(false);
+            closedCount++;
+        }
+
+        return closedCount;
+    }
+
+    public async Task ShutdownAsync(CancellationToken cancellationToken)
+    {
+        foreach (GatewaySession session in _registry.Snapshot())
+        {
+            try
+            {
+                await CloseSessionCoreAsync(session, GatewayShutdownReason, cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception exception)
+            {
+                _logger.LogWarning(
+                    exception,
+                    "Graceful shutdown failed for session {SessionId}; killing worker.",
+                    session.SessionId);
+                session.KillWorker(GatewayShutdownReason);
+            }
+        }
+    }
+
+    private async Task<SessionCloseResult> CloseSessionCoreAsync(
+        GatewaySession session,
+        string reason,
+        CancellationToken cancellationToken)
+    {
+        bool wasClosed = session.State == SessionState.Closed;
+        try
+        {
+            SessionCloseResult result = await session.CloseAsync(reason, cancellationToken).ConfigureAwait(false);
+            if (!wasClosed && !result.AlreadyClosed)
+            {
+                _metrics.SessionClosed();
+            }
+
+            return result;
+        }
+        catch (Exception exception)
+        {
+            session.MarkFaulted(exception.Message);
+            _metrics.Fault(SessionManagerErrorCode.CloseFailed.ToString());
+            throw new SessionManagerException(
+                SessionManagerErrorCode.CloseFailed,
+                $"Failed to close session {session.SessionId}.",
+                exception);
+        }
+    }
+
+    private GatewaySession GetRequiredSession(string sessionId)
+    {
+        if (!_registry.TryGet(sessionId, out GatewaySession session))
+        {
+            throw new SessionManagerException(
+                SessionManagerErrorCode.SessionNotFound,
+                $"Session {sessionId} was not found.");
+        }
+
+        return session;
+    }
+
+    private void EnsureSessionCapacity()
+    {
+        if (_registry.ActiveCount >= _options.Sessions.MaxSessions)
+        {
+            throw new SessionManagerException(
+                SessionManagerErrorCode.SessionLimitExceeded,
+                $"Gateway session limit {_options.Sessions.MaxSessions} has been reached.");
+        }
+    }
+
+    private GatewaySession CreateSession(
+        SessionOpenRequest request,
+        string? clientIdentity)
+    {
+        string sessionId = CreateSessionId();
+        string backendName = string.IsNullOrWhiteSpace(request.RequestedBackend)
+            ? GatewayContractInfo.DefaultBackendName
+            : request.RequestedBackend!;
+        TimeSpan commandTimeout = ResolveCommandTimeout(request.CommandTimeout);
+        TimeSpan startupTimeout = TimeSpan.FromSeconds(_options.Worker.StartupTimeoutSeconds);
+        TimeSpan shutdownTimeout = TimeSpan.FromSeconds(_options.Worker.ShutdownTimeoutSeconds);
+        string pipeName = $"mxaccess-gateway-{Environment.ProcessId}-{sessionId}";
+        string nonce = CreateNonce();
+        DateTimeOffset openedAt = _timeProvider.GetUtcNow();
+
+        return new GatewaySession(
+            sessionId,
+            backendName,
+            pipeName,
+            nonce,
+            clientIdentity,
+            request.ClientSessionName,
+            request.ClientCorrelationId,
+            commandTimeout,
+            startupTimeout,
+            shutdownTimeout,
+            openedAt);
+    }
+
+    private TimeSpan ResolveCommandTimeout(Duration? requestedTimeout)
+    {
+        if (requestedTimeout is null)
+        {
+            return TimeSpan.FromSeconds(_options.Sessions.DefaultCommandTimeoutSeconds);
+        }
+
+        TimeSpan timeout = requestedTimeout.ToTimeSpan();
+        return timeout <= TimeSpan.Zero
+            ? TimeSpan.FromSeconds(_options.Sessions.DefaultCommandTimeoutSeconds)
+            : timeout;
+    }
+
+    private static string CreateSessionId()
+    {
+        return $"session-{Guid.NewGuid():N}";
+    }
+
+    private static string CreateNonce()
+    {
+        Span<byte> bytes = stackalloc byte[32];
+        RandomNumberGenerator.Fill(bytes);
+
+        return Convert.ToBase64String(bytes);
+    }
+}

src/MxGateway.Server/Sessions/SessionManagerErrorCode.cs

@@ -0,0 +1,12 @@
+namespace MxGateway.Server.Sessions;
+
+public enum SessionManagerErrorCode
+{
+    SessionNotFound,
+    SessionNotReady,
+    EventSubscriberAlreadyActive,
+    EventQueueOverflow,
+    SessionLimitExceeded,
+    OpenFailed,
+    CloseFailed,
+}

src/MxGateway.Server/Sessions/SessionManagerException.cs

@@ -0,0 +1,23 @@
+namespace MxGateway.Server.Sessions;
+
+public sealed class SessionManagerException : Exception
+{
+    public SessionManagerException(
+        SessionManagerErrorCode errorCode,
+        string message)
+        : base(message)
+    {
+        ErrorCode = errorCode;
+    }
+
+    public SessionManagerException(
+        SessionManagerErrorCode errorCode,
+        string message,
+        Exception innerException)
+        : base(message, innerException)
+    {
+        ErrorCode = errorCode;
+    }
+
+    public SessionManagerErrorCode ErrorCode { get; }
+}

src/MxGateway.Server/Sessions/SessionOpenRequest.cs

@@ -0,0 +1,22 @@
+using Google.Protobuf.WellKnownTypes;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Server.Sessions;
+
+public sealed record SessionOpenRequest(
+    string? RequestedBackend,
+    string? ClientSessionName,
+    string? ClientCorrelationId,
+    Duration? CommandTimeout)
+{
+    public static SessionOpenRequest FromContract(OpenSessionRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+
+        return new SessionOpenRequest(
+            request.RequestedBackend,
+            request.ClientSessionName,
+            request.ClientCorrelationId,
+            request.CommandTimeout);
+    }
+}

src/MxGateway.Server/Sessions/SessionRegistry.cs

@@ -0,0 +1,39 @@
+using System.Collections.Concurrent;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Server.Sessions;
+
+public sealed class SessionRegistry : ISessionRegistry
+{
+    private readonly ConcurrentDictionary<string, GatewaySession> _sessions = new(StringComparer.Ordinal);
+
+    public int Count => _sessions.Count;
+
+    public int ActiveCount => _sessions.Values.Count(session => session.State is not SessionState.Closed);
+
+    public bool TryAdd(GatewaySession session)
+    {
+        ArgumentNullException.ThrowIfNull(session);
+
+        return _sessions.TryAdd(session.SessionId, session);
+    }
+
+    public bool TryGet(
+        string sessionId,
+        out GatewaySession session)
+    {
+        return _sessions.TryGetValue(sessionId, out session!);
+    }
+
+    public bool TryRemove(
+        string sessionId,
+        out GatewaySession session)
+    {
+        return _sessions.TryRemove(sessionId, out session!);
+    }
+
+    public IReadOnlyCollection<GatewaySession> Snapshot()
+    {
+        return _sessions.Values.ToArray();
+    }
+}

src/MxGateway.Server/Sessions/SessionServiceCollectionExtensions.cs

@@ -0,0 +1,13 @@
+namespace MxGateway.Server.Sessions;
+
+public static class SessionServiceCollectionExtensions
+{
+    public static IServiceCollection AddGatewaySessions(this IServiceCollection services)
+    {
+        services.AddSingleton<ISessionRegistry, SessionRegistry>();
+        services.AddSingleton<ISessionWorkerClientFactory, SessionWorkerClientFactory>();
+        services.AddSingleton<ISessionManager, SessionManager>();
+
+        return services;
+    }
+}

src/MxGateway.Server/Sessions/SessionWorkerClientFactory.cs

@@ -0,0 +1,144 @@
+using System.IO.Pipes;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using MxGateway.Contracts;
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Metrics;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Server.Sessions;
+
+public sealed class SessionWorkerClientFactory : ISessionWorkerClientFactory
+{
+    private readonly IWorkerProcessLauncher _workerProcessLauncher;
+    private readonly GatewayMetrics _metrics;
+    private readonly TimeProvider _timeProvider;
+    private readonly ILoggerFactory _loggerFactory;
+    private readonly GatewayOptions _options;
+
+    public SessionWorkerClientFactory(
+        IWorkerProcessLauncher workerProcessLauncher,
+        IOptions<GatewayOptions> options,
+        GatewayMetrics metrics,
+        ILoggerFactory loggerFactory,
+        TimeProvider? timeProvider = null)
+    {
+        _workerProcessLauncher = workerProcessLauncher ?? throw new ArgumentNullException(nameof(workerProcessLauncher));
+        ArgumentNullException.ThrowIfNull(options);
+        _metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
+        _loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
+        _timeProvider = timeProvider ?? TimeProvider.System;
+        _options = options.Value;
+    }
+
+    public async Task<IWorkerClient> CreateAsync(
+        GatewaySession session,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(session);
+
+        NamedPipeServerStream? pipe = CreatePipe(session.PipeName);
+        WorkerProcessHandle? processHandle = null;
+        IWorkerClient? workerClient = null;
+        try
+        {
+            session.TransitionTo(SessionState.StartingWorker);
+            processHandle = await _workerProcessLauncher
+                .LaunchAsync(
+                    new WorkerProcessLaunchRequest(
+                        session.SessionId,
+                        session.PipeName,
+                        GatewayContractInfo.WorkerProtocolVersion,
+                        session.Nonce,
+                        pipe),
+                    cancellationToken)
+                .ConfigureAwait(false);
+
+            session.TransitionTo(SessionState.WaitingForPipe);
+            await WaitForPipeConnectionAsync(pipe, session.StartupTimeout, cancellationToken).ConfigureAwait(false);
+
+            session.TransitionTo(SessionState.Handshaking);
+            WorkerFrameProtocolOptions frameOptions = new(
+                session.SessionId,
+                GatewayContractInfo.WorkerProtocolVersion,
+                _options.Worker.MaxMessageBytes);
+            WorkerClientConnection connection = new(
+                session.SessionId,
+                session.Nonce,
+                pipe,
+                frameOptions,
+                processHandle);
+            WorkerClientOptions clientOptions = new()
+            {
+                HeartbeatGrace = TimeSpan.FromSeconds(_options.Worker.HeartbeatGraceSeconds),
+                HeartbeatCheckInterval = TimeSpan.FromSeconds(_options.Worker.HeartbeatIntervalSeconds),
+                EventChannelCapacity = _options.Events.QueueCapacity,
+            };
+
+            workerClient = new WorkerClient(
+                connection,
+                clientOptions,
+                _metrics,
+                _timeProvider,
+                _loggerFactory.CreateLogger<WorkerClient>());
+
+            pipe = null;
+            processHandle = null;
+
+            session.TransitionTo(SessionState.InitializingWorker);
+            await workerClient.StartAsync(cancellationToken).ConfigureAwait(false);
+
+            return workerClient;
+        }
+        catch
+        {
+            if (workerClient is not null)
+            {
+                await workerClient.DisposeAsync().ConfigureAwait(false);
+            }
+            else
+            {
+                if (processHandle is not null)
+                {
+                    try
+                    {
+                        if (!processHandle.Process.HasExited)
+                        {
+                            processHandle.Process.Kill(entireProcessTree: true);
+                            _metrics.WorkerKilled("OpenSessionFailed");
+                        }
+                    }
+                    finally
+                    {
+                        processHandle.Dispose();
+                    }
+                }
+
+                pipe?.Dispose();
+            }
+
+            throw;
+        }
+    }
+
+    private static NamedPipeServerStream CreatePipe(string pipeName)
+    {
+        return new NamedPipeServerStream(
+            pipeName,
+            PipeDirection.InOut,
+            maxNumberOfServerInstances: 1,
+            PipeTransmissionMode.Byte,
+            PipeOptions.Asynchronous);
+    }
+
+    private static async Task WaitForPipeConnectionAsync(
+        NamedPipeServerStream pipe,
+        TimeSpan startupTimeout,
+        CancellationToken cancellationToken)
+    {
+        using CancellationTokenSource timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+        timeout.CancelAfter(startupTimeout);
+        await pipe.WaitForConnectionAsync(timeout.Token).ConfigureAwait(false);
+    }
+}

src/MxGateway.Server/Workers/WorkerClient.cs

@@ -29,6 +29,7 @@ public sealed class WorkerClient : IWorkerClient
     private WorkerClientState _state;
     private DateTimeOffset _lastHeartbeatAt;
     private int? _processId;
+    private int _eventQueueDepth;
     private Task? _readLoopTask;
     private Task? _writeLoopTask;
     private Task? _heartbeatLoopTask;
@@ -197,6 +198,8 @@ public sealed class WorkerClient : IWorkerClient
     {
         await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
         {
+            int queueDepth = Math.Max(0, Interlocked.Decrement(ref _eventQueueDepth));
+            _metrics?.SetEventQueueDepth(queueDepth);
             yield return workerEvent;
         }
     }
@@ -394,11 +397,6 @@ public sealed class WorkerClient : IWorkerClient
             _metrics?.EventReceived(SessionId, workerEvent.Event.Family.ToString());
         }
 
-        if (!await _events.Writer.WaitToWriteAsync(cancellationToken).ConfigureAwait(false))
-        {
-            return;
-        }
-
         if (!_events.Writer.TryWrite(workerEvent))
         {
             _metrics?.QueueOverflow("worker-events");
@@ -406,7 +404,11 @@ public sealed class WorkerClient : IWorkerClient
                 WorkerClientErrorCode.ProtocolViolation,
                 "Worker event channel rejected an event.",
                 null);
+            return;
         }
+
+        int queueDepth = Interlocked.Increment(ref _eventQueueDepth);
+        _metrics?.SetEventQueueDepth(queueDepth);
     }
 
     private void CompleteCommand(WorkerEnvelope envelope)

src/MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotServiceTests.cs

@@ -0,0 +1,290 @@
+using Microsoft.Extensions.Options;
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Dashboard;
+using MxGateway.Server.Metrics;
+using MxGateway.Server.Sessions;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Tests.Gateway.Dashboard;
+
+public sealed class DashboardSnapshotServiceTests
+{
+    [Fact]
+    public void GetSnapshot_WhenRegistryEmpty_ReturnsEmptyOperationalState()
+    {
+        using GatewayMetrics metrics = new();
+        DashboardSnapshotService service = CreateService(new SessionRegistry(), metrics);
+
+        DashboardSnapshot snapshot = service.GetSnapshot();
+
+        Assert.Empty(snapshot.Sessions);
+        Assert.Empty(snapshot.Workers);
+        Assert.Empty(snapshot.Faults);
+        Assert.Contains(snapshot.Metrics, metric => metric.Name == "mxgateway.sessions.open" && metric.Value == 0);
+        Assert.Equal("Healthy", snapshot.GatewayStatus);
+        Assert.NotNull(snapshot.Configuration);
+    }
+
+    [Fact]
+    public void GetSnapshot_ProjectsActiveAndFaultedSessionsWorkersMetricsAndFaults()
+    {
+        SessionRegistry registry = new();
+        GatewaySession activeSession = CreateSession(
+            "session-active",
+            "client-one",
+            DateTimeOffset.Parse("2026-04-26T10:00:00Z"));
+        activeSession.AttachWorkerClient(new FakeWorkerClient("session-active", 1201, WorkerClientState.Ready));
+        activeSession.MarkReady();
+        GatewaySession faultedSession = CreateSession(
+            "session-faulted",
+            "client-two",
+            DateTimeOffset.Parse("2026-04-26T10:01:00Z"));
+        faultedSession.AttachWorkerClient(new FakeWorkerClient("session-faulted", 1202, WorkerClientState.Faulted));
+        faultedSession.MarkFaulted("worker pipe disconnected");
+        registry.TryAdd(activeSession);
+        registry.TryAdd(faultedSession);
+        using GatewayMetrics metrics = new();
+        metrics.SessionOpened();
+        metrics.SessionOpened();
+        metrics.CommandStarted("Register");
+        metrics.CommandFailed("Register", "WorkerFaulted", TimeSpan.FromMilliseconds(7));
+        metrics.EventReceived("session-active", "OnDataChange");
+        metrics.Fault("WorkerFaulted");
+        DashboardSnapshotService service = CreateService(registry, metrics);
+
+        DashboardSnapshot snapshot = service.GetSnapshot();
+
+        Assert.Equal(2, snapshot.Sessions.Count);
+        Assert.Equal("session-faulted", snapshot.Sessions[0].SessionId);
+        Assert.Equal(SessionState.Faulted, snapshot.Sessions[0].State);
+        Assert.Equal(2, snapshot.Workers.Count);
+        Assert.Contains(snapshot.Metrics, metric => metric.Name == "mxgateway.commands.started" && metric.Value == 1);
+        Assert.Contains(
+            snapshot.Metrics,
+            metric => metric.Name == "mxgateway.events.received"
+                && metric.Dimension == "OnDataChange"
+                && metric.Value == 1);
+        DashboardFaultSummary fault = Assert.Single(snapshot.Faults);
+        Assert.Equal("Worker", fault.Source);
+        Assert.Equal("session-faulted", fault.SessionId);
+        Assert.Equal("worker pipe disconnected", fault.Message);
+    }
+
+    [Fact]
+    public void GetSnapshot_RedactsSecretsFromSessionAndFaultFields()
+    {
+        SessionRegistry registry = new();
+        GatewaySession session = CreateSession(
+            "session-redacted",
+            "Bearer mxgw_admin_super-secret",
+            DateTimeOffset.Parse("2026-04-26T10:00:00Z"),
+            clientSessionName: "password=hunter2",
+            clientCorrelationId: "token=abc123");
+        session.MarkFaulted("secret=credential-value");
+        registry.TryAdd(session);
+        using GatewayMetrics metrics = new();
+        DashboardSnapshotService service = CreateService(registry, metrics);
+
+        DashboardSnapshot snapshot = service.GetSnapshot();
+
+        DashboardSessionSummary summary = Assert.Single(snapshot.Sessions);
+        Assert.Equal("Bearer mxgw_admin_[redacted]", summary.ClientIdentity);
+        Assert.Equal("[redacted]", summary.ClientSessionName);
+        Assert.Equal("[redacted]", summary.ClientCorrelationId);
+        Assert.Equal("[redacted]", summary.LastFault);
+        Assert.Equal("[redacted]", Assert.Single(snapshot.Faults).Message);
+        Assert.Equal("[redacted]", snapshot.Configuration.Authentication.PepperSecretName);
+    }
+
+    [Fact]
+    public void GetSnapshot_DoesNotMutateSessionOrWorkerState()
+    {
+        SessionRegistry registry = new();
+        GatewaySession session = CreateSession(
+            "session-active",
+            "client-one",
+            DateTimeOffset.Parse("2026-04-26T10:00:00Z"));
+        FakeWorkerClient workerClient = new("session-active", 1201, WorkerClientState.Ready);
+        session.AttachWorkerClient(workerClient);
+        session.MarkReady();
+        registry.TryAdd(session);
+        using GatewayMetrics metrics = new();
+        DashboardSnapshotService service = CreateService(registry, metrics);
+
+        service.GetSnapshot();
+        service.GetSnapshot();
+
+        Assert.Equal(1, registry.ActiveCount);
+        Assert.Equal(SessionState.Ready, session.State);
+        Assert.Equal(WorkerClientState.Ready, workerClient.State);
+        Assert.Equal(0, workerClient.StartCount);
+        Assert.Equal(0, workerClient.ShutdownCount);
+        Assert.Equal(0, workerClient.KillCount);
+    }
+
+    [Fact]
+    public void GetSnapshot_AppliesRecentSessionAndFaultLimits()
+    {
+        SessionRegistry registry = new();
+        GatewaySession olderSession = CreateSession(
+            "session-older",
+            "client-one",
+            DateTimeOffset.Parse("2026-04-26T10:00:00Z"));
+        GatewaySession newerSession = CreateSession(
+            "session-newer",
+            "client-two",
+            DateTimeOffset.Parse("2026-04-26T10:01:00Z"));
+        olderSession.MarkFaulted("older fault");
+        newerSession.MarkFaulted("newer fault");
+        registry.TryAdd(olderSession);
+        registry.TryAdd(newerSession);
+        using GatewayMetrics metrics = new();
+        DashboardSnapshotService service = CreateService(
+            registry,
+            metrics,
+            new GatewayOptions
+            {
+                Dashboard = new DashboardOptions
+                {
+                    SnapshotIntervalMilliseconds = 1,
+                    RecentSessionLimit = 1,
+                    RecentFaultLimit = 1,
+                },
+            });
+
+        DashboardSnapshot snapshot = service.GetSnapshot();
+
+        Assert.Equal("session-newer", Assert.Single(snapshot.Sessions).SessionId);
+        Assert.Equal("session-newer", Assert.Single(snapshot.Faults).SessionId);
+    }
+
+    [Fact]
+    public async Task WatchSnapshotsAsync_WhenSubscriberCancels_DisposesCleanly()
+    {
+        using GatewayMetrics metrics = new();
+        DashboardSnapshotService service = CreateService(
+            new SessionRegistry(),
+            metrics,
+            new GatewayOptions
+            {
+                Dashboard = new DashboardOptions
+                {
+                    SnapshotIntervalMilliseconds = 1,
+                },
+            });
+        using CancellationTokenSource cancellation = new();
+        await using IAsyncEnumerator<DashboardSnapshot> enumerator = service
+            .WatchSnapshotsAsync(cancellation.Token)
+            .GetAsyncEnumerator();
+
+        Assert.True(await enumerator.MoveNextAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(1)));
+        await cancellation.CancelAsync();
+        bool hasNext = await enumerator.MoveNextAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(1));
+
+        Assert.False(hasNext);
+    }
+
+    private static DashboardSnapshotService CreateService(
+        SessionRegistry registry,
+        GatewayMetrics metrics,
+        GatewayOptions? options = null)
+    {
+        GatewayOptions resolvedOptions = options ?? new GatewayOptions
+        {
+            Dashboard = new DashboardOptions
+            {
+                SnapshotIntervalMilliseconds = 1,
+            },
+        };
+        GatewayConfigurationProvider configurationProvider = new(Options.Create(resolvedOptions));
+
+        return new DashboardSnapshotService(
+            registry,
+            metrics,
+            configurationProvider,
+            Options.Create(resolvedOptions));
+    }
+
+    private static GatewaySession CreateSession(
+        string sessionId,
+        string? clientIdentity,
+        DateTimeOffset openedAt,
+        string? clientSessionName = "test-session",
+        string? clientCorrelationId = "client-correlation")
+    {
+        return new GatewaySession(
+            sessionId,
+            "mxaccess",
+            $"mxaccess-gateway-1-{sessionId}",
+            "nonce",
+            clientIdentity,
+            clientSessionName,
+            clientCorrelationId,
+            TimeSpan.FromSeconds(30),
+            TimeSpan.FromSeconds(5),
+            TimeSpan.FromSeconds(5),
+            openedAt);
+    }
+
+    private sealed class FakeWorkerClient(
+        string sessionId,
+        int? processId,
+        WorkerClientState state) : IWorkerClient
+    {
+        public string SessionId { get; } = sessionId;
+
+        public int? ProcessId { get; } = processId;
+
+        public WorkerClientState State { get; private set; } = state;
+
+        public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.Parse("2026-04-26T10:02:00Z");
+
+        public int StartCount { get; private set; }
+
+        public int ShutdownCount { get; private set; }
+
+        public int KillCount { get; private set; }
+
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            StartCount++;
+            return Task.CompletedTask;
+        }
+
+        public Task<WorkerCommandReply> InvokeAsync(
+            WorkerCommand command,
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(new WorkerCommandReply());
+        }
+
+        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+            [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            await Task.CompletedTask;
+            yield break;
+        }
+
+        public Task ShutdownAsync(
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            ShutdownCount++;
+            State = WorkerClientState.Closed;
+            return Task.CompletedTask;
+        }
+
+        public void Kill(string reason)
+        {
+            KillCount++;
+            State = WorkerClientState.Faulted;
+        }
+
+        public ValueTask DisposeAsync()
+        {
+            return ValueTask.CompletedTask;
+        }
+    }
+}

src/MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs

@@ -0,0 +1,383 @@
+using System.Runtime.CompilerServices;
+using Grpc.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using MxGateway.Contracts;
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Grpc;
+using MxGateway.Server.Metrics;
+using MxGateway.Server.Sessions;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Tests.Gateway.Grpc;
+
+public sealed class EventStreamServiceTests
+{
+    private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);
+
+    [Fact]
+    public async Task StreamEventsAsync_YieldsEventsInWorkerOrder()
+    {
+        FakeWorkerClient workerClient = new();
+        GatewaySession session = CreateReadySession(workerClient);
+        FakeSessionManager sessionManager = new(session);
+        using GatewayMetrics metrics = new();
+        EventStreamService service = CreateService(sessionManager, metrics: metrics);
+        workerClient.Events.Add(CreateWorkerEvent(sequence: 10, MxEventFamily.OnDataChange));
+        workerClient.Events.Add(CreateWorkerEvent(sequence: 11, MxEventFamily.OnWriteComplete));
+        workerClient.CompleteAfterConfiguredEvents = true;
+
+        List<MxEvent> events = await CollectEventsAsync(service, session.SessionId);
+
+        Assert.Equal([10UL, 11UL], events.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
+        Assert.Equal(MxEventFamily.OnDataChange, events[0].Family);
+        Assert.Equal(MxEventFamily.OnWriteComplete, events[1].Family);
+        Assert.Equal(1, metrics.GetSnapshot().StreamDisconnects);
+    }
+
+    [Fact]
+    public async Task StreamEventsAsync_WhenSecondSubscriberStarts_RejectsClearly()
+    {
+        FakeWorkerClient workerClient = new();
+        GatewaySession session = CreateReadySession(workerClient);
+        EventStreamService service = CreateService(new FakeSessionManager(session));
+        using CancellationTokenSource firstSubscriberCancellation = new();
+        await using IAsyncEnumerator<MxEvent> firstSubscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), 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)
+            .GetAsyncEnumerator();
+
+        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
+            async () => await secondSubscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
+
+        Assert.Equal(SessionManagerErrorCode.EventSubscriberAlreadyActive, exception.ErrorCode);
+        await firstSubscriberCancellation.CancelAsync();
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(
+            async () => await firstMoveTask.WaitAsync(TestTimeout));
+        await firstSubscriber.DisposeAsync();
+        await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 0);
+    }
+
+    [Fact]
+    public async Task StreamEventsAsync_WhenCanceled_DetachesSubscriber()
+    {
+        FakeWorkerClient workerClient = new();
+        GatewaySession session = CreateReadySession(workerClient);
+        EventStreamService service = CreateService(new FakeSessionManager(session));
+        using CancellationTokenSource cancellationTokenSource = new();
+        await using IAsyncEnumerator<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), cancellationTokenSource.Token)
+            .GetAsyncEnumerator(cancellationTokenSource.Token);
+        Task<bool> moveTask = subscriber.MoveNextAsync().AsTask();
+
+        await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 1);
+        await cancellationTokenSource.CancelAsync();
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(
+            async () => await moveTask.WaitAsync(TestTimeout));
+        await subscriber.DisposeAsync();
+
+        await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 0);
+    }
+
+    [Fact]
+    public async Task StreamEventsAsync_WhenStreamQueueOverflows_FaultsSessionAndReportsOverflow()
+    {
+        FakeWorkerClient workerClient = new();
+        GatewaySession session = CreateReadySession(workerClient);
+        using GatewayMetrics metrics = new();
+        EventStreamService service = CreateService(
+            new FakeSessionManager(session),
+            metrics,
+            queueCapacity: 1);
+        workerClient.Events.Add(CreateWorkerEvent(sequence: 1, MxEventFamily.OnDataChange));
+        workerClient.Events.Add(CreateWorkerEvent(sequence: 2, MxEventFamily.OnDataChange));
+        workerClient.Events.Add(CreateWorkerEvent(sequence: 3, MxEventFamily.OnDataChange));
+        workerClient.CompleteAfterConfiguredEvents = true;
+        await using IAsyncEnumerator<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
+            .GetAsyncEnumerator();
+
+        Assert.True(await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
+        await WaitUntilAsync(() => session.State == SessionState.Faulted);
+        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
+            async () => await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
+
+        Assert.Equal(SessionManagerErrorCode.EventQueueOverflow, exception.ErrorCode);
+        Assert.Equal(SessionState.Faulted, session.State);
+        Assert.Equal(1, metrics.GetSnapshot().QueueOverflows);
+        Assert.Equal(1, metrics.GetSnapshot().Faults);
+    }
+
+    [Fact]
+    public async Task StreamEventsAsync_DoesNotSynthesizeOperationComplete()
+    {
+        FakeWorkerClient workerClient = new();
+        GatewaySession session = CreateReadySession(workerClient);
+        EventStreamService service = CreateService(new FakeSessionManager(session));
+        workerClient.Events.Add(CreateWorkerEvent(sequence: 10, MxEventFamily.OnWriteComplete));
+        workerClient.CompleteAfterConfiguredEvents = true;
+
+        List<MxEvent> events = await CollectEventsAsync(service, session.SessionId);
+
+        MxEvent mxEvent = Assert.Single(events);
+        Assert.Equal(MxEventFamily.OnWriteComplete, mxEvent.Family);
+        Assert.DoesNotContain(events, candidate => candidate.Family == MxEventFamily.OperationComplete);
+    }
+
+    [Fact]
+    public async Task StreamEventsAsync_WhenWorkerEventStreamFaults_PropagatesTerminalFault()
+    {
+        FakeWorkerClient workerClient = new()
+        {
+            TerminalException = new WorkerClientException(
+                WorkerClientErrorCode.WorkerFaulted,
+                "worker terminal fault"),
+        };
+        GatewaySession session = CreateReadySession(workerClient);
+        using GatewayMetrics metrics = new();
+        EventStreamService service = CreateService(new FakeSessionManager(session), metrics);
+        await using IAsyncEnumerator<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
+            .GetAsyncEnumerator();
+
+        WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
+            async () => await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
+
+        Assert.Equal(WorkerClientErrorCode.WorkerFaulted, exception.ErrorCode);
+        Assert.Equal(SessionState.Faulted, session.State);
+        Assert.Equal(1, metrics.GetSnapshot().Faults);
+    }
+
+    private static EventStreamService CreateService(
+        FakeSessionManager sessionManager,
+        GatewayMetrics? metrics = null,
+        int queueCapacity = 8)
+    {
+        return new EventStreamService(
+            sessionManager,
+            Options.Create(new GatewayOptions
+            {
+                Events = new EventOptions
+                {
+                    QueueCapacity = queueCapacity,
+                },
+            }),
+            new MxAccessGrpcMapper(),
+            metrics ?? new GatewayMetrics(),
+            NullLogger<EventStreamService>.Instance);
+    }
+
+    private static async Task<List<MxEvent>> CollectEventsAsync(
+        EventStreamService service,
+        string sessionId)
+    {
+        List<MxEvent> events = [];
+        await foreach (MxEvent mxEvent in service
+            .StreamEventsAsync(CreateRequest(sessionId), CancellationToken.None)
+            .WithCancellation(CancellationToken.None))
+        {
+            events.Add(mxEvent);
+        }
+
+        return events;
+    }
+
+    private static StreamEventsRequest CreateRequest(string sessionId)
+    {
+        return new StreamEventsRequest
+        {
+            SessionId = sessionId,
+        };
+    }
+
+    private static GatewaySession CreateReadySession(FakeWorkerClient workerClient)
+    {
+        GatewaySession session = new(
+            "session-events",
+            GatewayContractInfo.DefaultBackendName,
+            "pipe",
+            "nonce",
+            "client",
+            "client-session",
+            "client-correlation",
+            TimeSpan.FromSeconds(30),
+            TimeSpan.FromSeconds(30),
+            TimeSpan.FromSeconds(10),
+            DateTimeOffset.UtcNow);
+        session.AttachWorkerClient(workerClient);
+        session.MarkReady();
+
+        return session;
+    }
+
+    private static WorkerEvent CreateWorkerEvent(
+        ulong sequence,
+        MxEventFamily family)
+    {
+        MxEvent mxEvent = new()
+        {
+            SessionId = "session-events",
+            Family = family,
+            WorkerSequence = sequence,
+        };
+
+        switch (family)
+        {
+            case MxEventFamily.OnDataChange:
+                mxEvent.OnDataChange = new OnDataChangeEvent();
+                break;
+            case MxEventFamily.OnWriteComplete:
+                mxEvent.OnWriteComplete = new OnWriteCompleteEvent();
+                break;
+            case MxEventFamily.OperationComplete:
+                mxEvent.OperationComplete = new OperationCompleteEvent();
+                break;
+            case MxEventFamily.OnBufferedDataChange:
+                mxEvent.OnBufferedDataChange = new OnBufferedDataChangeEvent();
+                break;
+        }
+
+        return new WorkerEvent
+        {
+            Event = mxEvent,
+        };
+    }
+
+    private static async Task WaitUntilAsync(Func<bool> predicate)
+    {
+        using CancellationTokenSource cancellationTokenSource = new(TestTimeout);
+        while (!predicate())
+        {
+            await Task.Delay(TimeSpan.FromMilliseconds(10), cancellationTokenSource.Token);
+        }
+    }
+
+    private sealed class FakeSessionManager(GatewaySession session) : ISessionManager
+    {
+        public Task<GatewaySession> OpenSessionAsync(
+            SessionOpenRequest request,
+            string? clientIdentity,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(session);
+        }
+
+        public bool TryGetSession(
+            string sessionId,
+            out GatewaySession gatewaySession)
+        {
+            gatewaySession = session;
+            return string.Equals(sessionId, session.SessionId, StringComparison.Ordinal);
+        }
+
+        public Task<WorkerCommandReply> InvokeAsync(
+            string sessionId,
+            WorkerCommand command,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(new WorkerCommandReply());
+        }
+
+        public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+            string sessionId,
+            CancellationToken cancellationToken)
+        {
+            return session.ReadEventsAsync(cancellationToken);
+        }
+
+        public Task<SessionCloseResult> CloseSessionAsync(
+            string sessionId,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
+        }
+
+        public Task<int> CloseExpiredLeasesAsync(
+            DateTimeOffset now,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(0);
+        }
+
+        public Task ShutdownAsync(CancellationToken cancellationToken)
+        {
+            return Task.CompletedTask;
+        }
+    }
+
+    private sealed class FakeWorkerClient : IWorkerClient
+    {
+        public List<WorkerEvent> Events { get; } = [];
+
+        public bool CompleteAfterConfiguredEvents { get; set; }
+
+        public Exception? TerminalException { get; init; }
+
+        public string SessionId { get; } = "session-events";
+
+        public int? ProcessId { get; } = 4321;
+
+        public WorkerClientState State { get; private set; } = WorkerClientState.Ready;
+
+        public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
+
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            return Task.CompletedTask;
+        }
+
+        public Task<WorkerCommandReply> InvokeAsync(
+            WorkerCommand command,
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(new WorkerCommandReply());
+        }
+
+        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            foreach (WorkerEvent workerEvent in Events)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                yield return workerEvent;
+            }
+
+            if (TerminalException is not null)
+            {
+                throw TerminalException;
+            }
+
+            if (CompleteAfterConfiguredEvents)
+            {
+                yield break;
+            }
+
+            await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken);
+        }
+
+        public Task ShutdownAsync(
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            State = WorkerClientState.Closed;
+            return Task.CompletedTask;
+        }
+
+        public void Kill(string reason)
+        {
+            State = WorkerClientState.Faulted;
+        }
+
+        public ValueTask DisposeAsync()
+        {
+            return ValueTask.CompletedTask;
+        }
+    }
+}

src/MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs

@@ -0,0 +1,485 @@
+using System.Runtime.CompilerServices;
+using Google.Protobuf.WellKnownTypes;
+using Grpc.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using MxGateway.Contracts;
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Grpc;
+using MxGateway.Server.Security.Authentication;
+using MxGateway.Server.Security.Authorization;
+using MxGateway.Server.Sessions;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Tests.Gateway.Grpc;
+
+public sealed class MxAccessGatewayServiceTests
+{
+    [Fact]
+    public async Task OpenSession_WithValidRequest_ReturnsSessionDetails()
+    {
+        GatewayRequestIdentityAccessor identityAccessor = new();
+        FakeSessionManager sessionManager = new()
+        {
+            OpenSessionResult = CreateSession("session-1", processId: 4321),
+        };
+        MxAccessGatewayService service = CreateService(sessionManager, identityAccessor);
+
+        using IDisposable identityScope = identityAccessor.Push(CreateIdentity());
+        OpenSessionReply reply = await service.OpenSession(
+            new OpenSessionRequest
+            {
+                ClientSessionName = "operator-session",
+                CommandTimeout = Duration.FromTimeSpan(TimeSpan.FromSeconds(7)),
+            },
+            new TestServerCallContext());
+
+        Assert.Equal("session-1", reply.SessionId);
+        Assert.Equal(GatewayContractInfo.DefaultBackendName, reply.BackendName);
+        Assert.Equal(4321, reply.WorkerProcessId);
+        Assert.Equal(GatewayContractInfo.WorkerProtocolVersion, reply.WorkerProtocolVersion);
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Contains("unary-invoke", reply.Capabilities);
+        Assert.Equal("Operator Key", sessionManager.LastClientIdentity);
+        Assert.Equal("operator-session", sessionManager.LastOpenRequest?.ClientSessionName);
+    }
+
+    [Fact]
+    public async Task Invoke_WhenSessionMissing_ThrowsNotFound()
+    {
+        FakeSessionManager sessionManager = new()
+        {
+            InvokeException = new SessionManagerException(
+                SessionManagerErrorCode.SessionNotFound,
+                "Session session-missing was not found."),
+        };
+        MxAccessGatewayService service = CreateService(sessionManager);
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            async () => await service.Invoke(
+                CreatePingRequest("session-missing"),
+                new TestServerCallContext()));
+
+        Assert.Equal(StatusCode.NotFound, exception.StatusCode);
+        Assert.Contains("session-missing", exception.Status.Detail, StringComparison.Ordinal);
+    }
+
+    [Fact]
+    public async Task Invoke_WithMismatchedPayload_ThrowsInvalidArgumentAndDoesNotCallSessionManager()
+    {
+        FakeSessionManager sessionManager = new();
+        MxAccessGatewayService service = CreateService(sessionManager);
+        MxCommandRequest request = new()
+        {
+            SessionId = "session-1",
+            Command = new MxCommand
+            {
+                Kind = MxCommandKind.AddItem,
+                Ping = new PingCommand { Message = "wrong-payload" },
+            },
+        };
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            async () => await service.Invoke(request, new TestServerCallContext()));
+
+        Assert.Equal(StatusCode.InvalidArgument, exception.StatusCode);
+        Assert.Equal(0, sessionManager.InvokeCount);
+    }
+
+    [Fact]
+    public async Task Invoke_WithWorkerReply_ReturnsHresultStatusAndMethodPayload()
+    {
+        const int hresult = unchecked((int)0x80004005);
+        FakeSessionManager sessionManager = new()
+        {
+            InvokeReply = new WorkerCommandReply
+            {
+                Reply = new MxCommandReply
+                {
+                    SessionId = "session-1",
+                    CorrelationId = "worker-correlation",
+                    Kind = MxCommandKind.AddItem,
+                    ProtocolStatus = MxAccessGrpcMapper.Ok(),
+                    Hresult = hresult,
+                    AddItem = new AddItemReply { ItemHandle = 42 },
+                    DiagnosticMessage = "mxaccess diagnostic",
+                },
+            },
+        };
+        sessionManager.InvokeReply.Reply.Statuses.Add(new MxStatusProxy
+        {
+            Success = 0,
+            Category = MxStatusCategory.SoftwareError,
+            Detail = 1001,
+            DiagnosticText = "status detail",
+        });
+        MxAccessGatewayService service = CreateService(sessionManager);
+        MxCommandRequest request = new()
+        {
+            SessionId = "session-1",
+            ClientCorrelationId = "client-correlation",
+            Command = new MxCommand
+            {
+                Kind = MxCommandKind.AddItem,
+                AddItem = new AddItemCommand
+                {
+                    ServerHandle = 12,
+                    ItemDefinition = "Galaxy.Tag.Value",
+                },
+            },
+        };
+
+        MxCommandReply reply = await service.Invoke(request, new TestServerCallContext());
+
+        Assert.Equal(MxCommandKind.AddItem, sessionManager.LastWorkerCommand?.Command.Kind);
+        Assert.Equal("Galaxy.Tag.Value", sessionManager.LastWorkerCommand?.Command.AddItem.ItemDefinition);
+        Assert.NotNull(sessionManager.LastWorkerCommand?.EnqueueTimestamp);
+        Assert.Equal(hresult, reply.Hresult);
+        Assert.Equal(42, reply.AddItem.ItemHandle);
+        Assert.Equal("status detail", Assert.Single(reply.Statuses).DiagnosticText);
+        Assert.Equal("mxaccess diagnostic", reply.DiagnosticMessage);
+    }
+
+    [Fact]
+    public async Task StreamEvents_WithAfterSequence_WritesOnlyLaterEvents()
+    {
+        FakeSessionManager sessionManager = new();
+        sessionManager.Events.Add(CreateWorkerEvent("session-1", workerSequence: 1));
+        sessionManager.Events.Add(CreateWorkerEvent("session-1", workerSequence: 2));
+        MxAccessGatewayService service = CreateService(sessionManager);
+        TestServerStreamWriter<MxEvent> writer = new();
+
+        await service.StreamEvents(
+            new StreamEventsRequest
+            {
+                SessionId = "session-1",
+                AfterWorkerSequence = 1,
+            },
+            writer,
+            new TestServerCallContext());
+
+        MxEvent writtenEvent = Assert.Single(writer.Messages);
+        Assert.Equal((ulong)2, writtenEvent.WorkerSequence);
+        Assert.Equal("session-1", sessionManager.LastReadEventsSessionId);
+    }
+
+    [Fact]
+    public async Task CloseSession_WithBlankSessionId_ThrowsInvalidArgument()
+    {
+        MxAccessGatewayService service = CreateService(new FakeSessionManager());
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            async () => await service.CloseSession(
+                new CloseSessionRequest(),
+                new TestServerCallContext()));
+
+        Assert.Equal(StatusCode.InvalidArgument, exception.StatusCode);
+    }
+
+    private static MxAccessGatewayService CreateService(
+        FakeSessionManager sessionManager,
+        IGatewayRequestIdentityAccessor? identityAccessor = null)
+    {
+        return new MxAccessGatewayService(
+            sessionManager,
+            identityAccessor ?? new GatewayRequestIdentityAccessor(),
+            new MxAccessGrpcRequestValidator(),
+            new MxAccessGrpcMapper(),
+            new FakeEventStreamService(sessionManager),
+            NullLogger<MxAccessGatewayService>.Instance);
+    }
+
+    private static ApiKeyIdentity CreateIdentity()
+    {
+        return new ApiKeyIdentity(
+            KeyId: "operator01",
+            KeyPrefix: "mxgw_operator01",
+            DisplayName: "Operator Key",
+            Scopes: new HashSet<string>(StringComparer.Ordinal));
+    }
+
+    private static GatewaySession CreateSession(
+        string sessionId,
+        int processId)
+    {
+        GatewaySession session = new(
+            sessionId,
+            GatewayContractInfo.DefaultBackendName,
+            "pipe",
+            "nonce",
+            "Operator Key",
+            "operator-session",
+            "client-correlation",
+            TimeSpan.FromSeconds(7),
+            TimeSpan.FromSeconds(30),
+            TimeSpan.FromSeconds(10),
+            DateTimeOffset.UtcNow);
+        session.AttachWorkerClient(new FakeWorkerClient(processId));
+        session.MarkReady();
+
+        return session;
+    }
+
+    private static MxCommandRequest CreatePingRequest(string sessionId)
+    {
+        return new MxCommandRequest
+        {
+            SessionId = sessionId,
+            Command = new MxCommand
+            {
+                Kind = MxCommandKind.Ping,
+                Ping = new PingCommand { Message = "ping" },
+            },
+        };
+    }
+
+    private static WorkerEvent CreateWorkerEvent(
+        string sessionId,
+        ulong workerSequence)
+    {
+        return new WorkerEvent
+        {
+            Event = new MxEvent
+            {
+                Family = MxEventFamily.OnDataChange,
+                SessionId = sessionId,
+                WorkerSequence = workerSequence,
+                OnDataChange = new OnDataChangeEvent(),
+            },
+        };
+    }
+
+    private sealed class FakeSessionManager : ISessionManager
+    {
+        public GatewaySession? OpenSessionResult { get; init; }
+
+        public SessionOpenRequest? LastOpenRequest { get; private set; }
+
+        public string? LastClientIdentity { get; private set; }
+
+        public string? LastReadEventsSessionId { get; private set; }
+
+        public WorkerCommand? LastWorkerCommand { get; private set; }
+
+        public WorkerCommandReply InvokeReply { get; init; } = new()
+        {
+            Reply = new MxCommandReply
+            {
+                SessionId = "session-1",
+                Kind = MxCommandKind.Ping,
+                ProtocolStatus = MxAccessGrpcMapper.Ok(),
+            },
+        };
+
+        public Exception? InvokeException { get; init; }
+
+        public int InvokeCount { get; private set; }
+
+        public List<WorkerEvent> Events { get; } = [];
+
+        public void RecordReadEventsSessionId(string sessionId)
+        {
+            LastReadEventsSessionId = sessionId;
+        }
+
+        public Task<GatewaySession> OpenSessionAsync(
+            SessionOpenRequest request,
+            string? clientIdentity,
+            CancellationToken cancellationToken)
+        {
+            LastOpenRequest = request;
+            LastClientIdentity = clientIdentity;
+
+            return Task.FromResult(OpenSessionResult ?? CreateSession("session-1", processId: 1234));
+        }
+
+        public bool TryGetSession(
+            string sessionId,
+            out GatewaySession session)
+        {
+            session = OpenSessionResult ?? CreateSession(sessionId, processId: 1234);
+            return true;
+        }
+
+        public Task<WorkerCommandReply> InvokeAsync(
+            string sessionId,
+            WorkerCommand command,
+            CancellationToken cancellationToken)
+        {
+            InvokeCount++;
+            LastWorkerCommand = command;
+
+            if (InvokeException is not null)
+            {
+                throw InvokeException;
+            }
+
+            return Task.FromResult(InvokeReply);
+        }
+
+        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+            string sessionId,
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            LastReadEventsSessionId = sessionId;
+            foreach (WorkerEvent workerEvent in Events)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                await Task.Yield();
+                yield return workerEvent;
+            }
+        }
+
+        public Task<SessionCloseResult> CloseSessionAsync(
+            string sessionId,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
+        }
+
+        public Task<int> CloseExpiredLeasesAsync(
+            DateTimeOffset now,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(0);
+        }
+
+        public Task ShutdownAsync(CancellationToken cancellationToken)
+        {
+            return Task.CompletedTask;
+        }
+    }
+
+    private sealed class FakeEventStreamService(FakeSessionManager sessionManager) : IEventStreamService
+    {
+        public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
+            StreamEventsRequest request,
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            sessionManager.RecordReadEventsSessionId(request.SessionId);
+            foreach (WorkerEvent workerEvent in sessionManager.Events)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                await Task.Yield();
+                if (workerEvent.Event.WorkerSequence <= request.AfterWorkerSequence)
+                {
+                    continue;
+                }
+
+                yield return workerEvent.Event;
+            }
+        }
+    }
+
+    private sealed class FakeWorkerClient(int processId) : IWorkerClient
+    {
+        public string SessionId { get; } = "session-1";
+
+        public int? ProcessId { get; } = processId;
+
+        public WorkerClientState State { get; } = WorkerClientState.Ready;
+
+        public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
+
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            return Task.CompletedTask;
+        }
+
+        public Task<WorkerCommandReply> InvokeAsync(
+            WorkerCommand command,
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(new WorkerCommandReply());
+        }
+
+        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            await Task.CompletedTask;
+            yield break;
+        }
+
+        public Task ShutdownAsync(
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            return Task.CompletedTask;
+        }
+
+        public void Kill(string reason)
+        {
+        }
+
+        public ValueTask DisposeAsync()
+        {
+            return ValueTask.CompletedTask;
+        }
+    }
+
+    private sealed class TestServerStreamWriter<T> : IServerStreamWriter<T>
+    {
+        public List<T> Messages { get; } = [];
+
+        public WriteOptions? WriteOptions { get; set; }
+
+        public Task WriteAsync(T message)
+        {
+            Messages.Add(message);
+
+            return Task.CompletedTask;
+        }
+    }
+
+    private sealed class TestServerCallContext(CancellationToken cancellationToken = default) : ServerCallContext
+    {
+        private readonly Metadata requestHeaders = [];
+        private readonly Metadata responseTrailers = [];
+        private readonly Dictionary<object, object> userState = [];
+        private Status status;
+        private WriteOptions? writeOptions;
+
+        protected override string MethodCore => "/mxaccess_gateway.v1.MxAccessGateway/Test";
+
+        protected override string HostCore => "localhost";
+
+        protected override string PeerCore => "ipv4:127.0.0.1:5000";
+
+        protected override DateTime DeadlineCore => DateTime.UtcNow.AddMinutes(1);
+
+        protected override Metadata RequestHeadersCore => requestHeaders;
+
+        protected override CancellationToken CancellationTokenCore => cancellationToken;
+
+        protected override Metadata ResponseTrailersCore => responseTrailers;
+
+        protected override Status StatusCore
+        {
+            get => status;
+            set => status = value;
+        }
+
+        protected override WriteOptions? WriteOptionsCore
+        {
+            get => writeOptions;
+            set => writeOptions = value;
+        }
+
+        protected override AuthContext AuthContextCore { get; } = new(
+            string.Empty,
+            new Dictionary<string, List<AuthProperty>>(StringComparer.Ordinal));
+
+        protected override IDictionary<object, object> UserStateCore => userState;
+
+        protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders)
+        {
+            return Task.CompletedTask;
+        }
+
+        protected override ContextPropagationToken CreatePropagationTokenCore(
+            ContextPropagationOptions? options)
+        {
+            throw new NotSupportedException();
+        }
+    }
+}

src/MxGateway.Tests/Gateway/Grpc/MxAccessGrpcMapperTests.cs

@@ -0,0 +1,76 @@
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Grpc;
+
+namespace MxGateway.Tests.Gateway.Grpc;
+
+public sealed class MxAccessGrpcMapperTests
+{
+    [Fact]
+    public void MapCommand_ClonesMethodSpecificPayloadForWorkerBoundary()
+    {
+        MxAccessGrpcMapper mapper = new();
+        MxCommandRequest request = new()
+        {
+            SessionId = "session-1",
+            Command = new MxCommand
+            {
+                Kind = MxCommandKind.Write,
+                Write = new WriteCommand
+                {
+                    ServerHandle = 10,
+                    ItemHandle = 20,
+                    UserId = 30,
+                    Value = new MxValue
+                    {
+                        DataType = MxDataType.String,
+                        StringValue = "value",
+                    },
+                },
+            },
+        };
+
+        WorkerCommand workerCommand = mapper.MapCommand(request);
+        request.Command.Write.Value.StringValue = "changed";
+
+        Assert.Equal(MxCommandKind.Write, workerCommand.Command.Kind);
+        Assert.Equal("value", workerCommand.Command.Write.Value.StringValue);
+        Assert.NotNull(workerCommand.EnqueueTimestamp);
+    }
+
+    [Fact]
+    public void MapCommandReply_PreservesHresultStatusesAndPayload()
+    {
+        const int hresult = unchecked((int)0x80070005);
+        WorkerCommandReply workerReply = new()
+        {
+            Reply = new MxCommandReply
+            {
+                SessionId = "session-1",
+                Kind = MxCommandKind.Register,
+                ProtocolStatus = MxAccessGrpcMapper.Ok(),
+                Hresult = hresult,
+                Register = new RegisterReply { ServerHandle = 50 },
+            },
+        };
+        workerReply.Reply.Statuses.Add(new MxStatusProxy
+        {
+            Success = 0,
+            Category = MxStatusCategory.SecurityError,
+            DiagnosticText = "denied",
+        });
+
+        MxCommandReply publicReply = new MxAccessGrpcMapper().MapCommandReply(workerReply);
+
+        Assert.Equal(hresult, publicReply.Hresult);
+        Assert.Equal(50, publicReply.Register.ServerHandle);
+        Assert.Equal("denied", Assert.Single(publicReply.Statuses).DiagnosticText);
+    }
+
+    [Fact]
+    public void MapCommandReply_WhenWorkerReplyMissing_ReturnsProtocolViolationReply()
+    {
+        MxCommandReply publicReply = new MxAccessGrpcMapper().MapCommandReply(new WorkerCommandReply());
+
+        Assert.Equal(ProtocolStatusCode.ProtocolViolation, publicReply.ProtocolStatus.Code);
+    }
+}

src/MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs

@@ -0,0 +1,320 @@
+using Google.Protobuf.WellKnownTypes;
+using Microsoft.Extensions.Options;
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Metrics;
+using MxGateway.Server.Sessions;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Tests.Gateway.Sessions;
+
+public sealed class SessionManagerTests
+{
+    [Fact]
+    public async Task OpenSessionAsync_WithWorkerReady_RegistersReadySession()
+    {
+        FakeWorkerClient workerClient = new();
+        FakeSessionWorkerClientFactory factory = new(workerClient)
+        {
+            ApplyLifecycleTransitions = true,
+        };
+        using GatewayMetrics metrics = new();
+        SessionManager manager = CreateManager(factory, metrics: metrics);
+
+        GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", CancellationToken.None);
+
+        Assert.True(manager.TryGetSession(session.SessionId, out GatewaySession registered));
+        Assert.Same(session, registered);
+        Assert.Equal(SessionState.Ready, session.State);
+        Assert.Equal("client-1", session.ClientIdentity);
+        Assert.Equal(["StartingWorker", "WaitingForPipe", "Handshaking", "InitializingWorker"], factory.ObservedStates);
+        Assert.Equal(1, metrics.GetSnapshot().OpenSessions);
+        Assert.Equal(1, metrics.GetSnapshot().SessionsOpened);
+    }
+
+    [Fact]
+    public async Task InvokeAsync_WhenSessionReady_ForwardsCommandToWorker()
+    {
+        FakeWorkerClient workerClient = new();
+        SessionManager manager = CreateManager(new FakeSessionWorkerClientFactory(workerClient));
+        GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", CancellationToken.None);
+
+        WorkerCommandReply reply = await manager.InvokeAsync(
+            session.SessionId,
+            CreateCommand(MxCommandKind.Ping),
+            CancellationToken.None);
+
+        Assert.Equal(1, workerClient.InvokeCount);
+        Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind);
+    }
+
+    [Fact]
+    public async Task InvokeAsync_WhenSessionFaulted_RejectsCommand()
+    {
+        FakeWorkerClient workerClient = new();
+        SessionManager manager = CreateManager(new FakeSessionWorkerClientFactory(workerClient));
+        GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", CancellationToken.None);
+        session.MarkFaulted("test fault");
+
+        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
+            async () => await manager.InvokeAsync(
+                session.SessionId,
+                CreateCommand(MxCommandKind.Ping),
+                CancellationToken.None));
+
+        Assert.Equal(SessionManagerErrorCode.SessionNotReady, exception.ErrorCode);
+        Assert.Equal(0, workerClient.InvokeCount);
+    }
+
+    [Fact]
+    public async Task CloseSessionAsync_WhenCalledTwice_IsIdempotent()
+    {
+        FakeWorkerClient workerClient = new();
+        using GatewayMetrics metrics = new();
+        SessionManager manager = CreateManager(new FakeSessionWorkerClientFactory(workerClient), metrics: metrics);
+        GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", CancellationToken.None);
+
+        SessionCloseResult firstClose = await manager.CloseSessionAsync(session.SessionId, CancellationToken.None);
+        SessionCloseResult secondClose = await manager.CloseSessionAsync(session.SessionId, CancellationToken.None);
+
+        Assert.False(firstClose.AlreadyClosed);
+        Assert.True(secondClose.AlreadyClosed);
+        Assert.Equal(SessionState.Closed, firstClose.FinalState);
+        Assert.Equal(SessionState.Closed, secondClose.FinalState);
+        Assert.Equal(1, workerClient.ShutdownCount);
+        Assert.Equal(1, metrics.GetSnapshot().SessionsClosed);
+        Assert.Equal(0, metrics.GetSnapshot().OpenSessions);
+    }
+
+    [Fact]
+    public async Task OpenSessionAsync_WhenWorkerCreationFails_RemovesSessionFromRegistry()
+    {
+        SessionRegistry registry = new();
+        using GatewayMetrics metrics = new();
+        SessionManager manager = CreateManager(
+            new FailingSessionWorkerClientFactory(),
+            registry,
+            metrics);
+
+        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
+            async () => await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", CancellationToken.None));
+
+        Assert.Equal(SessionManagerErrorCode.OpenFailed, exception.ErrorCode);
+        Assert.Equal(0, registry.Count);
+        Assert.Equal(0, metrics.GetSnapshot().SessionsOpened);
+        Assert.Equal(1, metrics.GetSnapshot().Faults);
+    }
+
+    [Fact]
+    public async Task CloseExpiredLeasesAsync_ClosesExpiredSessionsOnly()
+    {
+        FakeWorkerClient expiredClient = new();
+        FakeWorkerClient activeClient = new();
+        QueueingSessionWorkerClientFactory factory = new(expiredClient, activeClient);
+        SessionManager manager = CreateManager(factory);
+        GatewaySession expiredSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", CancellationToken.None);
+        GatewaySession activeSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", CancellationToken.None);
+        DateTimeOffset now = DateTimeOffset.UtcNow;
+        expiredSession.ExtendLease(now.AddSeconds(-1));
+        activeSession.ExtendLease(now.AddMinutes(5));
+
+        int closedCount = await manager.CloseExpiredLeasesAsync(now, CancellationToken.None);
+
+        Assert.Equal(1, closedCount);
+        Assert.Equal(SessionState.Closed, expiredSession.State);
+        Assert.Equal(SessionState.Ready, activeSession.State);
+        Assert.Equal(1, expiredClient.ShutdownCount);
+        Assert.Equal(0, activeClient.ShutdownCount);
+    }
+
+    [Fact]
+    public async Task ShutdownAsync_ClosesAllRegisteredSessions()
+    {
+        FakeWorkerClient firstClient = new();
+        FakeWorkerClient secondClient = new();
+        QueueingSessionWorkerClientFactory factory = new(firstClient, secondClient);
+        using GatewayMetrics metrics = new();
+        SessionManager manager = CreateManager(factory, metrics: metrics);
+        GatewaySession firstSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", CancellationToken.None);
+        GatewaySession secondSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", CancellationToken.None);
+
+        await manager.ShutdownAsync(CancellationToken.None);
+
+        Assert.Equal(SessionState.Closed, firstSession.State);
+        Assert.Equal(SessionState.Closed, secondSession.State);
+        Assert.Equal(1, firstClient.ShutdownCount);
+        Assert.Equal(1, secondClient.ShutdownCount);
+        Assert.Equal(2, metrics.GetSnapshot().SessionsClosed);
+        Assert.Equal(0, metrics.GetSnapshot().OpenSessions);
+    }
+
+    private static SessionManager CreateManager(
+        ISessionWorkerClientFactory factory,
+        ISessionRegistry? registry = null,
+        GatewayMetrics? metrics = null,
+        GatewayOptions? options = null)
+    {
+        return new SessionManager(
+            registry ?? new SessionRegistry(),
+            factory,
+            Options.Create(options ?? CreateOptions()),
+            metrics ?? new GatewayMetrics());
+    }
+
+    private static GatewayOptions CreateOptions()
+    {
+        return new GatewayOptions
+        {
+            Sessions = new SessionOptions
+            {
+                DefaultCommandTimeoutSeconds = 30,
+                MaxSessions = 64,
+            },
+            Worker = new WorkerOptions
+            {
+                StartupTimeoutSeconds = 30,
+                ShutdownTimeoutSeconds = 10,
+            },
+        };
+    }
+
+    private static SessionOpenRequest CreateOpenRequest()
+    {
+        return new SessionOpenRequest(
+            RequestedBackend: null,
+            ClientSessionName: "test-session",
+            ClientCorrelationId: "client-correlation-1",
+            CommandTimeout: Duration.FromTimeSpan(TimeSpan.FromSeconds(5)));
+    }
+
+    private static WorkerCommand CreateCommand(MxCommandKind kind)
+    {
+        return new WorkerCommand
+        {
+            Command = new MxCommand
+            {
+                Kind = kind,
+            },
+        };
+    }
+
+    private sealed class FakeSessionWorkerClientFactory(IWorkerClient workerClient) : ISessionWorkerClientFactory
+    {
+        public List<string> ObservedStates { get; } = [];
+
+        public bool ApplyLifecycleTransitions { get; init; }
+
+        public Task<IWorkerClient> CreateAsync(
+            GatewaySession session,
+            CancellationToken cancellationToken)
+        {
+            ObservedStates.Add(session.State.ToString());
+            if (ApplyLifecycleTransitions)
+            {
+                session.TransitionTo(SessionState.WaitingForPipe);
+                ObservedStates.Add(session.State.ToString());
+                session.TransitionTo(SessionState.Handshaking);
+                ObservedStates.Add(session.State.ToString());
+                session.TransitionTo(SessionState.InitializingWorker);
+                ObservedStates.Add(session.State.ToString());
+            }
+
+            return Task.FromResult(workerClient);
+        }
+    }
+
+    private sealed class QueueingSessionWorkerClientFactory : ISessionWorkerClientFactory
+    {
+        private readonly Queue<IWorkerClient> _workerClients;
+
+        public QueueingSessionWorkerClientFactory(params IWorkerClient[] workerClients)
+        {
+            _workerClients = new Queue<IWorkerClient>(workerClients);
+        }
+
+        public Task<IWorkerClient> CreateAsync(
+            GatewaySession session,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(_workerClients.Dequeue());
+        }
+    }
+
+    private sealed class FailingSessionWorkerClientFactory : ISessionWorkerClientFactory
+    {
+        public Task<IWorkerClient> CreateAsync(
+            GatewaySession session,
+            CancellationToken cancellationToken)
+        {
+            throw new InvalidOperationException("worker startup failed");
+        }
+    }
+
+    private sealed class FakeWorkerClient : IWorkerClient
+    {
+        public string SessionId { get; init; } = "session-1";
+
+        public int? ProcessId { get; init; } = 1234;
+
+        public WorkerClientState State { get; set; } = WorkerClientState.Ready;
+
+        public DateTimeOffset LastHeartbeatAt { get; init; } = DateTimeOffset.UtcNow;
+
+        public int InvokeCount { get; private set; }
+
+        public int ShutdownCount { get; private set; }
+
+        public int KillCount { get; private set; }
+
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            return Task.CompletedTask;
+        }
+
+        public Task<WorkerCommandReply> InvokeAsync(
+            WorkerCommand command,
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            InvokeCount++;
+            MxCommandKind kind = command.Command?.Kind ?? MxCommandKind.Unspecified;
+
+            return Task.FromResult(new WorkerCommandReply
+            {
+                Reply = new MxCommandReply
+                {
+                    SessionId = SessionId,
+                    CorrelationId = "correlation-1",
+                    Kind = kind,
+                },
+            });
+        }
+
+        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+            [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            await Task.CompletedTask;
+            yield break;
+        }
+
+        public Task ShutdownAsync(
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            ShutdownCount++;
+            State = WorkerClientState.Closed;
+            return Task.CompletedTask;
+        }
+
+        public void Kill(string reason)
+        {
+            KillCount++;
+            State = WorkerClientState.Faulted;
+        }
+
+        public ValueTask DisposeAsync()
+        {
+            return ValueTask.CompletedTask;
+        }
+    }
+}

src/MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs

@@ -109,6 +109,32 @@ public sealed class WorkerClientTests
         Assert.Equal(MxEventFamily.OperationComplete, events.Current.Event.Family);
     }
 
+    [Fact]
+    public async Task ReadLoop_WhenEventQueueOverflows_FaultsClient()
+    {
+        await using PipePair pipePair = await PipePair.CreateAsync();
+        await using WorkerClient client = CreateClient(
+            pipePair,
+            new WorkerClientOptions
+            {
+                EventChannelCapacity = 1,
+                HeartbeatGrace = TimeSpan.FromSeconds(30),
+                HeartbeatCheckInterval = TimeSpan.FromSeconds(30),
+            });
+        await CompleteHandshakeAsync(client, pipePair);
+
+        await pipePair.WorkerWriter.WriteAsync(
+            CreateEventEnvelope(sequence: 11, MxEventFamily.OnDataChange));
+        await pipePair.WorkerWriter.WriteAsync(
+            CreateEventEnvelope(sequence: 12, MxEventFamily.OnDataChange));
+
+        await WaitUntilAsync(
+            () => client.State == WorkerClientState.Faulted,
+            TestTimeout);
+
+        Assert.Equal(WorkerClientState.Faulted, client.State);
+    }
+
     [Fact]
     public async Task ReadLoop_WhenPipeDisconnects_FaultsClient()
     {

src/MxGateway.Worker.Tests/Conversion/HResultConverterTests.cs

@@ -0,0 +1,47 @@
+using System;
+using System.Runtime.InteropServices;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.Conversion;
+
+namespace MxGateway.Worker.Tests.Conversion;
+
+public sealed class HResultConverterTests
+{
+    private readonly HResultConverter _converter = new();
+
+    [Fact]
+    public void Convert_WithComException_CapturesExceptionHResult()
+    {
+        COMException exception = new("Sensitive provider text should not be copied.", unchecked((int)0x80070057));
+
+        HResultConversion converted = _converter.Convert(exception);
+
+        Assert.Equal(unchecked((int)0x80070057), converted.HResult);
+        Assert.Equal(ProtocolStatusCode.MxaccessFailure, converted.ProtocolStatus.Code);
+        Assert.Contains("0x80070057", converted.ProtocolStatus.Message);
+        Assert.Contains(typeof(COMException).FullName!, converted.DiagnosticMessage);
+        Assert.DoesNotContain("Sensitive provider text", converted.DiagnosticMessage);
+    }
+
+    [Fact]
+    public void CreateProtocolStatus_WithSuccessHResult_ReturnsOk()
+    {
+        ProtocolStatus status = _converter.CreateProtocolStatus(0);
+
+        Assert.Equal(ProtocolStatusCode.Ok, status.Code);
+        Assert.Equal("HRESULT 0x00000000", status.Message);
+    }
+
+    [Fact]
+    public void Convert_WithNonComException_CapturesExceptionHResult()
+    {
+        InvalidOperationException exception = new("do not include this");
+
+        HResultConversion converted = _converter.Convert(exception);
+
+        Assert.Equal(exception.HResult, converted.HResult);
+        Assert.Equal(ProtocolStatusCode.MxaccessFailure, converted.ProtocolStatus.Code);
+        Assert.Contains("0x", converted.DiagnosticMessage);
+        Assert.DoesNotContain("do not include this", converted.DiagnosticMessage);
+    }
+}

src/MxGateway.Worker.Tests/Conversion/MxStatusProxyConverterTests.cs

@@ -0,0 +1,118 @@
+using System;
+using System.Collections.Generic;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.Conversion;
+
+namespace MxGateway.Worker.Tests.Conversion;
+
+public sealed class MxStatusProxyConverterTests
+{
+    private readonly MxStatusProxyConverter _converter = new();
+
+    [Fact]
+    public void Convert_WithStatusStruct_PreservesStatusFields()
+    {
+        FakeMxStatusProxy status = new()
+        {
+            success = 1,
+            category = 5,
+            detectedBy = 3,
+            detail = 21,
+        };
+
+        MxStatusProxy converted = _converter.Convert(status);
+
+        Assert.Equal(1, converted.Success);
+        Assert.Equal(MxStatusCategory.OperationalError, converted.Category);
+        Assert.Equal(MxStatusSource.RespondingNmx, converted.DetectedBy);
+        Assert.Equal(21, converted.Detail);
+        Assert.Equal(5, converted.RawCategory);
+        Assert.Equal(3, converted.RawDetectedBy);
+        Assert.Equal("Invalid reference", converted.DiagnosticText);
+    }
+
+    [Fact]
+    public void ConvertMany_WithStatusArray_DoesNotCollapseEntries()
+    {
+        FakeMxStatusProxy[] statuses =
+        [
+            new()
+            {
+                success = 1,
+                category = 0,
+                detectedBy = 0,
+                detail = 0,
+            },
+            new()
+            {
+                success = 0,
+                category = 6,
+                detectedBy = 5,
+                detail = 33,
+            },
+        ];
+
+        IReadOnlyList<MxStatusProxy> converted = _converter.ConvertMany(statuses);
+
+        Assert.Equal(2, converted.Count);
+        Assert.Equal(MxStatusCategory.Ok, converted[0].Category);
+        Assert.Equal(MxStatusCategory.SecurityError, converted[1].Category);
+        Assert.Equal(MxStatusSource.RespondingAutomationObject, converted[1].DetectedBy);
+        Assert.Equal("Write access denied", converted[1].DiagnosticText);
+    }
+
+    [Fact]
+    public void Convert_WithUnknownCategoryAndSource_PreservesRawFields()
+    {
+        FakeMxStatusProxy status = new()
+        {
+            success = -1,
+            category = 99,
+            detectedBy = 42,
+            detail = 1234,
+        };
+
+        MxStatusProxy converted = _converter.Convert(status);
+
+        Assert.Equal(-1, converted.Success);
+        Assert.Equal(MxStatusCategory.Unknown, converted.Category);
+        Assert.Equal(MxStatusSource.Unknown, converted.DetectedBy);
+        Assert.Equal(99, converted.RawCategory);
+        Assert.Equal(42, converted.RawDetectedBy);
+        Assert.Equal(1234, converted.Detail);
+        Assert.Equal(string.Empty, converted.DiagnosticText);
+    }
+
+    [Fact]
+    public void PreserveCompletionOnlyStatusBytes_ReturnsRawHexMetadata()
+    {
+        string rawStatus = _converter.PreserveCompletionOnlyStatusBytes(
+            [0x00, 0x00, 0x50, 0x80, 0x00]);
+
+        Assert.Equal("completion_only_status_hex=0000508000", rawStatus);
+    }
+
+    [Fact]
+    public void Convert_WithMissingStatusField_ThrowsConversionException()
+    {
+        MxStatusConversionException exception =
+            Assert.Throws<MxStatusConversionException>(() => _converter.Convert(new MissingFields()));
+
+        Assert.Contains("success", exception.Message);
+    }
+
+    public struct FakeMxStatusProxy
+    {
+        public short success;
+
+        public int category;
+
+        public int detectedBy;
+
+        public short detail;
+    }
+
+    private sealed class MissingFields
+    {
+    }
+}

src/MxGateway.Worker.Tests/Conversion/VariantConverterTests.cs

@@ -0,0 +1,183 @@
+using System;
+using Google.Protobuf;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.Bootstrap;
+using MxGateway.Worker.Conversion;
+using ProtobufTimestamp = Google.Protobuf.WellKnownTypes.Timestamp;
+
+namespace MxGateway.Worker.Tests.Conversion;
+
+public sealed class VariantConverterTests
+{
+    private readonly VariantConverter _converter = new();
+
+    [Theory]
+    [InlineData(true, MxDataType.Boolean, MxValue.KindOneofCase.BoolValue)]
+    [InlineData(42, MxDataType.Integer, MxValue.KindOneofCase.Int32Value)]
+    [InlineData(42L, MxDataType.Integer, MxValue.KindOneofCase.Int64Value)]
+    [InlineData(1.25f, MxDataType.Float, MxValue.KindOneofCase.FloatValue)]
+    [InlineData(2.5d, MxDataType.Double, MxValue.KindOneofCase.DoubleValue)]
+    [InlineData("value", MxDataType.String, MxValue.KindOneofCase.StringValue)]
+    public void Convert_WithSupportedScalar_ProjectsTypedValue(
+        object value,
+        MxDataType expectedDataType,
+        MxValue.KindOneofCase expectedKind)
+    {
+        MxValue converted = _converter.Convert(value);
+
+        Assert.Equal(expectedDataType, converted.DataType);
+        Assert.Equal(expectedKind, converted.KindCase);
+        Assert.False(string.IsNullOrWhiteSpace(converted.VariantType));
+    }
+
+    [Fact]
+    public void Convert_WithDateTime_ProjectsTimestamp()
+    {
+        DateTime dateTime = new(2026, 4, 26, 17, 45, 0, DateTimeKind.Utc);
+
+        MxValue converted = _converter.Convert(dateTime);
+
+        Assert.Equal(MxDataType.Time, converted.DataType);
+        Assert.Equal(ProtobufTimestamp.FromDateTime(dateTime), converted.TimestampValue);
+        Assert.Equal("VT_DATE", converted.VariantType);
+    }
+
+    [Fact]
+    public void Convert_WithFileTimeAndExpectedTime_ProjectsTimestamp()
+    {
+        DateTime dateTime = new(2026, 4, 26, 17, 45, 0, DateTimeKind.Utc);
+
+        MxValue converted = _converter.Convert(dateTime.ToFileTimeUtc(), MxDataType.Time);
+
+        Assert.Equal(MxDataType.Time, converted.DataType);
+        Assert.Equal(ProtobufTimestamp.FromDateTime(dateTime), converted.TimestampValue);
+        Assert.Equal("VT_I8", converted.VariantType);
+    }
+
+    [Theory]
+    [InlineData(null, "VT_EMPTY")]
+    [InlineData(typeof(DBNull), "VT_NULL")]
+    public void Convert_WithNullLikeValue_PreservesNull(
+        object? value,
+        string expectedVariantType)
+    {
+        object? actualValue = value is System.Type ? DBNull.Value : value;
+
+        MxValue converted = _converter.Convert(actualValue);
+
+        Assert.True(converted.IsNull);
+        Assert.Equal(MxDataType.NoData, converted.DataType);
+        Assert.Equal(expectedVariantType, converted.VariantType);
+        Assert.Equal(MxValue.KindOneofCase.None, converted.KindCase);
+    }
+
+    [Fact]
+    public void ConvertArray_WithSupportedArrays_ProjectsTypedValuesAndDimensions()
+    {
+        MxValue bools = _converter.Convert(new[] { true, false });
+        MxValue ints = _converter.Convert(new[] { 1, 2, 3 });
+        MxValue floats = _converter.Convert(new[] { 1.25f, 2.5f });
+        MxValue doubles = _converter.Convert(new[] { 1.25d, 2.5d });
+        MxValue strings = _converter.Convert(new[] { "one", "two" });
+        MxValue times = _converter.Convert(new[]
+        {
+            new DateTime(2026, 4, 26, 17, 45, 0, DateTimeKind.Utc),
+            new DateTime(2026, 4, 26, 17, 46, 0, DateTimeKind.Utc),
+        });
+
+        Assert.Equal(new[] { true, false }, bools.ArrayValue.BoolValues.Values);
+        Assert.Equal(new[] { 1, 2, 3 }, ints.ArrayValue.Int32Values.Values);
+        Assert.Equal(new[] { 1.25f, 2.5f }, floats.ArrayValue.FloatValues.Values);
+        Assert.Equal(new[] { 1.25d, 2.5d }, doubles.ArrayValue.DoubleValues.Values);
+        Assert.Equal(new[] { "one", "two" }, strings.ArrayValue.StringValues.Values);
+        Assert.Equal(2, times.ArrayValue.TimestampValues.Values.Count);
+        Assert.Equal(new uint[] { 2 }, bools.ArrayValue.Dimensions);
+        Assert.Equal(MxDataType.Boolean, bools.ArrayValue.ElementDataType);
+    }
+
+    [Fact]
+    public void ConvertArray_WithMultidimensionalArray_PreservesRankAndDimensions()
+    {
+        int[,] values =
+        {
+            { 1, 2, 3 },
+            { 4, 5, 6 },
+        };
+
+        MxValue converted = _converter.Convert(values);
+
+        Assert.Equal(new uint[] { 2, 3 }, converted.ArrayValue.Dimensions);
+        Assert.Equal(new[] { 1, 2, 3, 4, 5, 6 }, converted.ArrayValue.Int32Values.Values);
+    }
+
+    [Fact]
+    public void ConvertArray_WithExpectedTimeAndFileTimeValues_ProjectsTimestampArray()
+    {
+        DateTime first = new(2026, 4, 26, 17, 45, 0, DateTimeKind.Utc);
+        DateTime second = new(2026, 4, 26, 17, 46, 0, DateTimeKind.Utc);
+
+        MxValue converted = _converter.Convert(
+            new[] { first.ToFileTimeUtc(), second.ToFileTimeUtc() },
+            MxDataType.Time);
+
+        Assert.Equal(MxDataType.Time, converted.ArrayValue.ElementDataType);
+        Assert.Equal(
+            new[] { ProtobufTimestamp.FromDateTime(first), ProtobufTimestamp.FromDateTime(second) },
+            converted.ArrayValue.TimestampValues.Values);
+    }
+
+    [Fact]
+    public void Convert_WithUnknownScalar_PreservesRawMetadata()
+    {
+        UnsupportedVariant value = new("opaque");
+
+        MxValue converted = _converter.Convert(value);
+
+        Assert.Equal(MxDataType.Unknown, converted.DataType);
+        Assert.Equal(MxValue.KindOneofCase.RawValue, converted.KindCase);
+        Assert.Contains(typeof(UnsupportedVariant).FullName!, converted.VariantType);
+        Assert.Contains(typeof(UnsupportedVariant).FullName!, converted.RawDiagnostic);
+        Assert.Equal(ByteString.CopyFromUtf8("opaque"), converted.RawValue);
+    }
+
+    [Fact]
+    public void ConvertArray_WithUnknownArray_PreservesRawMetadata()
+    {
+        UnsupportedVariant[] values =
+        [
+            new("first"),
+            new("second"),
+        ];
+
+        MxValue converted = _converter.Convert(values);
+
+        Assert.Equal(MxDataType.Unknown, converted.ArrayValue.ElementDataType);
+        Assert.Equal(MxArray.ValuesOneofCase.RawValues, converted.ArrayValue.ValuesCase);
+        Assert.Equal(new uint[] { 2 }, converted.ArrayValue.Dimensions);
+        Assert.Equal("first", converted.ArrayValue.RawValues.Values[0].ToStringUtf8());
+        Assert.Contains(typeof(UnsupportedVariant).FullName!, converted.ArrayValue.RawDiagnostic);
+    }
+
+    [Fact]
+    public void Redactor_WithCredentialBearingValueFields_RedactsBeforeLogging()
+    {
+        Assert.Equal(WorkerLogRedactor.RedactedValue, WorkerLogRedactor.RedactValue("credential_value", "secret"));
+        Assert.Equal(WorkerLogRedactor.RedactedValue, WorkerLogRedactor.RedactValue("password_value", "secret"));
+        Assert.Equal(WorkerLogRedactor.RedactedValue, WorkerLogRedactor.RedactValue("secured_write_token", "secret"));
+    }
+
+    private sealed class UnsupportedVariant
+    {
+        private readonly string _value;
+
+        public UnsupportedVariant(string value)
+        {
+            _value = value;
+        }
+
+        public override string ToString()
+        {
+            return _value;
+        }
+    }
+}

src/MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs

@@ -1,5 +1,6 @@
 using System.Collections.Generic;
 using System.IO;
+using System.Runtime.InteropServices;
 using System.Threading;
 using System.Threading.Tasks;
 using MxGateway.Contracts;
@@ -37,6 +38,10 @@ public sealed class WorkerPipeSessionTests
         Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerHello, written[0].BodyCase);
         Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerReady, written[1].BodyCase);
         Assert.Equal(Nonce, written[0].WorkerHello.Nonce);
+        Assert.Equal(1234, written[1].WorkerReady.WorkerProcessId);
+        Assert.Equal(MxGateway.Worker.MxAccess.MxAccessInteropInfo.ProgId, written[1].WorkerReady.MxaccessProgid);
+        Assert.Equal(MxGateway.Worker.MxAccess.MxAccessInteropInfo.Clsid, written[1].WorkerReady.MxaccessClsid);
+        Assert.NotNull(written[1].WorkerReady.ReadyTimestamp);
     }
 
     [Fact]
@@ -117,6 +122,31 @@ public sealed class WorkerPipeSessionTests
         Assert.Equal(WorkerFaultCategory.ProtocolViolation, fault.WorkerFault.Category);
     }
 
+    [Fact]
+    public async Task CompleteStartupHandshakeAsync_WhenMxAccessCreationFails_WritesFaultInsteadOfReady()
+    {
+        const int hresult = unchecked((int)0x80040154);
+        WorkerFrameProtocolOptions options = CreateOptions();
+        MemoryStream inbound = new();
+        await new WorkerFrameWriter(inbound, options).WriteAsync(CreateGatewayHelloEnvelope());
+        inbound.Position = 0;
+        MemoryStream outbound = new();
+        WorkerPipeSession session = CreateSession(inbound, outbound, options);
+
+        await Assert.ThrowsAsync<COMException>(
+            async () => await session.CompleteStartupHandshakeAsync(
+                _ => Task.FromException<WorkerReady>(new COMException("Class not registered.", hresult))));
+
+        WorkerEnvelope[] written = ReadWrittenFrames(outbound, options);
+        Assert.Equal(2, written.Length);
+        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerHello, written[0].BodyCase);
+        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerFault, written[1].BodyCase);
+        Assert.Equal(WorkerFaultCategory.MxaccessCreationFailed, written[1].WorkerFault.Category);
+        Assert.Equal(hresult, written[1].WorkerFault.Hresult);
+        Assert.Equal(typeof(COMException).FullName, written[1].WorkerFault.ExceptionType);
+        Assert.Equal(ProtocolStatusCode.WorkerUnavailable, written[1].WorkerFault.ProtocolStatus.Code);
+    }
+
     private static WorkerPipeSession CreateSession(
         Stream inbound,
         Stream outbound,

src/MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs

@@ -0,0 +1,220 @@
+using System;
+using System.Runtime.InteropServices;
+using System.Threading.Tasks;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.MxAccess;
+using MxGateway.Worker.Sta;
+
+namespace MxGateway.Worker.Tests.MxAccess;
+
+public sealed class MxAccessCommandExecutorTests
+{
+    [Fact]
+    public async Task DispatchAsync_Register_CallsMxAccessOnStaAndPreservesServerHandle()
+    {
+        FakeMxAccessComObjectFactory factory = new(new FakeMxAccessComObject(registerHandle: 42));
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+
+        MxCommandReply reply = await session.DispatchAsync(CreateRegisterCommand("correlation-1", "client-a"));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.True(reply.HasHresult);
+        Assert.Equal(0, reply.Hresult);
+        Assert.Equal(42, reply.Register.ServerHandle);
+        Assert.Equal(MxDataType.Integer, reply.ReturnValue.DataType);
+        Assert.Equal(42, reply.ReturnValue.Int32Value);
+        Assert.Equal(runtime.StaThreadId, factory.FakeComObject.RegisterThreadId);
+        Assert.Equal("client-a", factory.FakeComObject.RegisteredClientName);
+
+        RegisteredServerHandle registeredServerHandle = Assert.Single(
+            await session.GetRegisteredServerHandlesAsync());
+        Assert.Equal(42, registeredServerHandle.ServerHandle);
+        Assert.Equal("client-a", registeredServerHandle.ClientName);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_Unregister_CallsMxAccessOnStaAndRemovesTrackedServerHandle()
+    {
+        FakeMxAccessComObject fakeComObject = new(registerHandle: 43);
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register", "client-a"));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateUnregisterCommand("unregister", 43));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Equal(43, fakeComObject.UnregisteredServerHandle);
+        Assert.Equal(runtime.StaThreadId, fakeComObject.UnregisterThreadId);
+        Assert.Empty(await session.GetRegisteredServerHandlesAsync());
+    }
+
+    [Fact]
+    public async Task DispatchAsync_UnregisterWhenMxAccessThrows_PreservesHResultAndDoesNotRewriteFailure()
+    {
+        const int hresult = unchecked((int)0x80070057);
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 44,
+            unregisterException: new COMException("Invalid handle.", hresult));
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-failure", "client-a"));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateUnregisterCommand("invalid-unregister", 44));
+
+        Assert.Equal(ProtocolStatusCode.MxaccessFailure, reply.ProtocolStatus.Code);
+        Assert.True(reply.HasHresult);
+        Assert.Equal(hresult, reply.Hresult);
+        Assert.Contains("0x80070057", reply.DiagnosticMessage);
+        Assert.Equal(44, fakeComObject.UnregisteredServerHandle);
+
+        RegisteredServerHandle registeredServerHandle = Assert.Single(
+            await session.GetRegisteredServerHandlesAsync());
+        Assert.Equal(44, registeredServerHandle.ServerHandle);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_RegisterWithoutPayload_ReturnsInvalidRequest()
+    {
+        FakeMxAccessComObjectFactory factory = new(new FakeMxAccessComObject(registerHandle: 45));
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+
+        MxCommandReply reply = await session.DispatchAsync(new StaCommand(
+            "session-1",
+            "missing-payload",
+            new MxCommand
+            {
+                Kind = MxCommandKind.Register,
+            }));
+
+        Assert.Equal(ProtocolStatusCode.InvalidRequest, reply.ProtocolStatus.Code);
+        Assert.Null(factory.FakeComObject.RegisteredClientName);
+    }
+
+    private static StaCommand CreateRegisterCommand(
+        string correlationId,
+        string clientName)
+    {
+        return new StaCommand(
+            "session-1",
+            correlationId,
+            new MxCommand
+            {
+                Kind = MxCommandKind.Register,
+                Register = new RegisterCommand
+                {
+                    ClientName = clientName,
+                },
+            });
+    }
+
+    private static StaCommand CreateUnregisterCommand(
+        string correlationId,
+        int serverHandle)
+    {
+        return new StaCommand(
+            "session-1",
+            correlationId,
+            new MxCommand
+            {
+                Kind = MxCommandKind.Unregister,
+                Unregister = new UnregisterCommand
+                {
+                    ServerHandle = serverHandle,
+                },
+            });
+    }
+
+    private static StaRuntime CreateRuntime()
+    {
+        return new StaRuntime(
+            new NoopComApartmentInitializer(),
+            new StaMessagePump(),
+            TimeSpan.FromMilliseconds(25));
+    }
+
+    private sealed class FakeMxAccessComObject
+    {
+        private readonly int registerHandle;
+        private readonly Exception? unregisterException;
+
+        public FakeMxAccessComObject(
+            int registerHandle,
+            Exception? unregisterException = null)
+        {
+            this.registerHandle = registerHandle;
+            this.unregisterException = unregisterException;
+        }
+
+        public string? RegisteredClientName { get; private set; }
+
+        public int? RegisterThreadId { get; private set; }
+
+        public int? UnregisteredServerHandle { get; private set; }
+
+        public int? UnregisterThreadId { get; private set; }
+
+        public int Register(string clientName)
+        {
+            RegisteredClientName = clientName;
+            RegisterThreadId = Environment.CurrentManagedThreadId;
+
+            return registerHandle;
+        }
+
+        public void Unregister(int serverHandle)
+        {
+            UnregisteredServerHandle = serverHandle;
+            UnregisterThreadId = Environment.CurrentManagedThreadId;
+
+            if (unregisterException is not null)
+            {
+                throw unregisterException;
+            }
+        }
+    }
+
+    private sealed class FakeMxAccessComObjectFactory : IMxAccessComObjectFactory
+    {
+        public FakeMxAccessComObjectFactory(FakeMxAccessComObject fakeComObject)
+        {
+            FakeComObject = fakeComObject;
+        }
+
+        public FakeMxAccessComObject FakeComObject { get; }
+
+        public object Create()
+        {
+            return FakeComObject;
+        }
+    }
+
+    private sealed class NoopEventSink : IMxAccessEventSink
+    {
+        public void Attach(object mxAccessComObject)
+        {
+        }
+
+        public void Detach()
+        {
+        }
+    }
+
+    private sealed class NoopComApartmentInitializer : IStaComApartmentInitializer
+    {
+        public void Initialize()
+        {
+        }
+
+        public void Uninitialize()
+        {
+        }
+    }
+}

src/MxGateway.Worker.Tests/MxAccess/MxAccessLiveComCreationTests.cs

@@ -0,0 +1,75 @@
+using System;
+using System.Threading.Tasks;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.MxAccess;
+using MxGateway.Worker.Sta;
+
+namespace MxGateway.Worker.Tests.MxAccess;
+
+public sealed class MxAccessLiveComCreationTests
+{
+    [Fact]
+    public async Task StartAsync_WhenOptedIn_CreatesInstalledMxAccessComObjectOnSta()
+    {
+        if (!string.Equals(
+                Environment.GetEnvironmentVariable("MXGATEWAY_RUN_LIVE_MXACCESS_TESTS"),
+                "1",
+                StringComparison.Ordinal))
+        {
+            return;
+        }
+
+        using MxAccessStaSession session = new();
+
+        await session.StartAsync(workerProcessId: 1234);
+    }
+
+    [Fact]
+    public async Task RegisterAndUnregister_WhenOptedIn_RoundTripsInstalledMxAccessServerHandle()
+    {
+        if (!RunLiveMxAccessTests())
+        {
+            return;
+        }
+
+        using MxAccessStaSession session = new();
+        await session.StartAsync(workerProcessId: 1234);
+
+        MxCommandReply registerReply = await session.DispatchAsync(new StaCommand(
+            "session-1",
+            "live-register",
+            new MxCommand
+            {
+                Kind = MxCommandKind.Register,
+                Register = new RegisterCommand
+                {
+                    ClientName = "MxGateway.Worker.Tests",
+                },
+            }));
+
+        Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code);
+        Assert.True(registerReply.Register.ServerHandle > 0);
+
+        MxCommandReply unregisterReply = await session.DispatchAsync(new StaCommand(
+            "session-1",
+            "live-unregister",
+            new MxCommand
+            {
+                Kind = MxCommandKind.Unregister,
+                Unregister = new UnregisterCommand
+                {
+                    ServerHandle = registerReply.Register.ServerHandle,
+                },
+            }));
+
+        Assert.Equal(ProtocolStatusCode.Ok, unregisterReply.ProtocolStatus.Code);
+    }
+
+    private static bool RunLiveMxAccessTests()
+    {
+        return string.Equals(
+            Environment.GetEnvironmentVariable("MXGATEWAY_RUN_LIVE_MXACCESS_TESTS"),
+            "1",
+            StringComparison.Ordinal);
+    }
+}

src/MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs

@@ -0,0 +1,133 @@
+using System;
+using System.Runtime.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.MxAccess;
+using MxGateway.Worker.Sta;
+
+namespace MxGateway.Worker.Tests.MxAccess;
+
+public sealed class MxAccessStaSessionTests
+{
+    [Fact]
+    public async Task StartAsync_CreatesComObjectAndAttachesEventSinkOnStaThread()
+    {
+        FakeMxAccessComObjectFactory factory = new();
+        FakeMxAccessEventSink eventSink = new();
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, eventSink);
+
+        WorkerReady ready = await session.StartAsync(workerProcessId: 1234);
+
+        Assert.Equal(1234, ready.WorkerProcessId);
+        Assert.Equal(MxAccessInteropInfo.ProgId, ready.MxaccessProgid);
+        Assert.Equal(MxAccessInteropInfo.Clsid, ready.MxaccessClsid);
+        Assert.NotNull(ready.ReadyTimestamp);
+        Assert.Equal(runtime.StaThreadId, factory.CreateThreadId);
+        Assert.Equal(runtime.StaThreadId, eventSink.AttachThreadId);
+        Assert.Equal(ApartmentState.STA, factory.CreateApartmentState);
+        Assert.Same(factory.CreatedObject, eventSink.AttachedObject);
+    }
+
+    [Fact]
+    public async Task StartAsync_WhenFactoryFails_MapsCreationExceptionWithHResult()
+    {
+        const int hresult = unchecked((int)0x80040154);
+        FakeMxAccessComObjectFactory factory = new(new COMException("Class not registered.", hresult));
+        FakeMxAccessEventSink eventSink = new();
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, eventSink);
+
+        MxAccessCreationException exception = await Assert.ThrowsAsync<MxAccessCreationException>(
+            () => session.StartAsync(workerProcessId: 1234));
+
+        Assert.Equal(hresult, exception.CapturedHResult);
+        Assert.Equal(MxAccessInteropInfo.ProgId, exception.AttemptedProgId);
+        Assert.Equal(MxAccessInteropInfo.Clsid, exception.AttemptedClsid);
+        Assert.Null(eventSink.AttachedObject);
+    }
+
+    [Fact]
+    public async Task Dispose_DetachesEventSinkOnStaThread()
+    {
+        FakeMxAccessComObjectFactory factory = new();
+        FakeMxAccessEventSink eventSink = new();
+        using StaRuntime runtime = CreateRuntime();
+        MxAccessStaSession session = new(runtime, factory, eventSink);
+        await session.StartAsync(workerProcessId: 1234);
+
+        session.Dispose();
+
+        Assert.Equal(runtime.StaThreadId, eventSink.DetachThreadId);
+    }
+
+    private static StaRuntime CreateRuntime()
+    {
+        return new StaRuntime(
+            new NoopComApartmentInitializer(),
+            new StaMessagePump(),
+            TimeSpan.FromMilliseconds(25));
+    }
+
+    private sealed class FakeMxAccessComObjectFactory : IMxAccessComObjectFactory
+    {
+        private readonly Exception? exception;
+
+        public FakeMxAccessComObjectFactory(Exception? exception = null)
+        {
+            this.exception = exception;
+        }
+
+        public object CreatedObject { get; } = new();
+
+        public int? CreateThreadId { get; private set; }
+
+        public ApartmentState? CreateApartmentState { get; private set; }
+
+        public object Create()
+        {
+            CreateThreadId = Thread.CurrentThread.ManagedThreadId;
+            CreateApartmentState = Thread.CurrentThread.GetApartmentState();
+
+            if (exception is not null)
+            {
+                throw exception;
+            }
+
+            return CreatedObject;
+        }
+    }
+
+    private sealed class FakeMxAccessEventSink : IMxAccessEventSink
+    {
+        public object? AttachedObject { get; private set; }
+
+        public int? AttachThreadId { get; private set; }
+
+        public int? DetachThreadId { get; private set; }
+
+        public void Attach(object mxAccessComObject)
+        {
+            AttachedObject = mxAccessComObject;
+            AttachThreadId = Thread.CurrentThread.ManagedThreadId;
+        }
+
+        public void Detach()
+        {
+            DetachThreadId = Thread.CurrentThread.ManagedThreadId;
+            AttachedObject = null;
+        }
+    }
+
+    private sealed class NoopComApartmentInitializer : IStaComApartmentInitializer
+    {
+        public void Initialize()
+        {
+        }
+
+        public void Uninitialize()
+        {
+        }
+    }
+}

src/MxGateway.Worker.Tests/Sta/StaCommandDispatcherTests.cs

@@ -0,0 +1,279 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Runtime.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.Sta;
+
+namespace MxGateway.Worker.Tests.Sta;
+
+public sealed class StaCommandDispatcherTests
+{
+    [Fact]
+    public async Task DispatchAsync_ExecutesCommandsOnStaInQueueOrder()
+    {
+        using StaRuntime runtime = CreateRuntime();
+        runtime.Start();
+        RecordingCommandExecutor executor = new();
+        StaCommandDispatcher dispatcher = new(runtime, executor);
+
+        Task<MxCommandReply> first = dispatcher.DispatchAsync(CreateCommand("correlation-1", MxCommandKind.Register));
+        Task<MxCommandReply> second = dispatcher.DispatchAsync(CreateCommand("correlation-2", MxCommandKind.AddItem));
+
+        MxCommandReply[] replies = await Task.WhenAll(first, second);
+
+        Assert.Equal(new[] { "correlation-1", "correlation-2" }, executor.CorrelationIds);
+        Assert.All(executor.ThreadIds, threadId => Assert.Equal(runtime.StaThreadId, threadId));
+        Assert.Equal("correlation-1", replies[0].CorrelationId);
+        Assert.Equal("correlation-2", replies[1].CorrelationId);
+        Assert.Equal(ProtocolStatusCode.Ok, replies[0].ProtocolStatus.Code);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_WhenExecutorThrows_ReturnsFailureReplyWithHResult()
+    {
+        using StaRuntime runtime = CreateRuntime();
+        runtime.Start();
+        StaCommandDispatcher dispatcher = new(
+            runtime,
+            new ThrowingCommandExecutor(new COMException("provider detail", unchecked((int)0x80070057))));
+
+        MxCommandReply reply = await dispatcher.DispatchAsync(CreateCommand("correlation-1", MxCommandKind.Register));
+
+        Assert.Equal("session-1", reply.SessionId);
+        Assert.Equal("correlation-1", reply.CorrelationId);
+        Assert.Equal(MxCommandKind.Register, reply.Kind);
+        Assert.Equal(ProtocolStatusCode.MxaccessFailure, reply.ProtocolStatus.Code);
+        Assert.Equal(unchecked((int)0x80070057), reply.Hresult);
+        Assert.Contains("0x80070057", reply.DiagnosticMessage);
+        Assert.DoesNotContain("provider detail", reply.DiagnosticMessage);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_WhenCanceledBeforeExecution_ReturnsCanceledReplyWithoutExecuting()
+    {
+        using StaRuntime runtime = CreateRuntime();
+        runtime.Start();
+        BlockingCommandExecutor executor = new();
+        StaCommandDispatcher dispatcher = new(runtime, executor);
+        Task<MxCommandReply> blocked = dispatcher.DispatchAsync(CreateCommand("blocked", MxCommandKind.Register));
+        Assert.True(executor.Started.Wait(TimeSpan.FromSeconds(2)));
+
+        using CancellationTokenSource cancellation = new();
+        Task<MxCommandReply> canceled = dispatcher.DispatchAsync(
+            CreateCommand("canceled", MxCommandKind.AddItem, cancellation.Token));
+        cancellation.Cancel();
+
+        executor.Release();
+        MxCommandReply canceledReply = await canceled;
+        await blocked;
+
+        Assert.Equal(ProtocolStatusCode.Canceled, canceledReply.ProtocolStatus.Code);
+        Assert.DoesNotContain("canceled", executor.CorrelationIds);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_WhenCanceledAfterExecutionStarts_StillReturnsLateReply()
+    {
+        using StaRuntime runtime = CreateRuntime();
+        runtime.Start();
+        BlockingCommandExecutor executor = new();
+        StaCommandDispatcher dispatcher = new(runtime, executor);
+        using CancellationTokenSource cancellation = new();
+
+        Task<MxCommandReply> replyTask = dispatcher.DispatchAsync(
+            CreateCommand("late-reply", MxCommandKind.Register, cancellation.Token));
+
+        Assert.True(executor.Started.Wait(TimeSpan.FromSeconds(2)));
+        cancellation.Cancel();
+        executor.Release();
+
+        MxCommandReply reply = await replyTask;
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Contains("late-reply", executor.CorrelationIds);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_WhenShutdownRequested_RejectsNewCommands()
+    {
+        using StaRuntime runtime = CreateRuntime();
+        runtime.Start();
+        StaCommandDispatcher dispatcher = new(runtime, new RecordingCommandExecutor());
+
+        dispatcher.RequestShutdown();
+        MxCommandReply reply = await dispatcher.DispatchAsync(CreateCommand("correlation-1", MxCommandKind.Register));
+
+        Assert.Equal(ProtocolStatusCode.WorkerUnavailable, reply.ProtocolStatus.Code);
+        Assert.Equal("correlation-1", reply.CorrelationId);
+    }
+
+    [Fact]
+    public async Task PopulateHeartbeat_ReportsCurrentCorrelationAndPendingCount()
+    {
+        using StaRuntime runtime = CreateRuntime();
+        runtime.Start();
+        BlockingCommandExecutor executor = new();
+        StaCommandDispatcher dispatcher = new(runtime, executor);
+
+        Task<MxCommandReply> current = dispatcher.DispatchAsync(CreateCommand("current", MxCommandKind.Register));
+        Assert.True(executor.Started.Wait(TimeSpan.FromSeconds(2)));
+        Task<MxCommandReply> pending = dispatcher.DispatchAsync(CreateCommand("pending", MxCommandKind.AddItem));
+
+        WorkerHeartbeat heartbeat = new();
+        dispatcher.PopulateHeartbeat(heartbeat);
+
+        Assert.Equal("current", heartbeat.CurrentCommandCorrelationId);
+        Assert.Equal(1u, heartbeat.PendingCommandCount);
+
+        executor.Release();
+        await Task.WhenAll(current, pending);
+    }
+
+    private static StaCommand CreateCommand(
+        string correlationId,
+        MxCommandKind kind,
+        CancellationToken cancellationToken = default)
+    {
+        return new StaCommand(
+            "session-1",
+            correlationId,
+            new MxCommand
+            {
+                Kind = kind,
+                Ping = new PingCommand
+                {
+                    Message = correlationId,
+                },
+            },
+            cancellationToken: cancellationToken);
+    }
+
+    private static StaRuntime CreateRuntime()
+    {
+        return new StaRuntime(
+            new NoopComApartmentInitializer(),
+            new StaMessagePump(),
+            TimeSpan.FromMilliseconds(25));
+    }
+
+    private sealed class RecordingCommandExecutor : IStaCommandExecutor
+    {
+        private readonly object gate = new();
+        private readonly List<string> correlationIds = new();
+        private readonly List<int> threadIds = new();
+
+        public IReadOnlyList<string> CorrelationIds
+        {
+            get
+            {
+                lock (gate)
+                {
+                    return correlationIds.ToArray();
+                }
+            }
+        }
+
+        public IReadOnlyList<int> ThreadIds
+        {
+            get
+            {
+                lock (gate)
+                {
+                    return threadIds.ToArray();
+                }
+            }
+        }
+
+        public MxCommandReply Execute(StaCommand command)
+        {
+            lock (gate)
+            {
+                correlationIds.Add(command.CorrelationId);
+                threadIds.Add(Thread.CurrentThread.ManagedThreadId);
+            }
+
+            return new MxCommandReply
+            {
+                ProtocolStatus = new ProtocolStatus
+                {
+                    Code = ProtocolStatusCode.Ok,
+                    Message = "OK",
+                },
+            };
+        }
+    }
+
+    private sealed class BlockingCommandExecutor : IStaCommandExecutor
+    {
+        private readonly ManualResetEventSlim release = new(false);
+        private readonly object gate = new();
+        private readonly List<string> correlationIds = new();
+
+        public ManualResetEventSlim Started { get; } = new(false);
+
+        public IReadOnlyList<string> CorrelationIds
+        {
+            get
+            {
+                lock (gate)
+                {
+                    return correlationIds.ToArray();
+                }
+            }
+        }
+
+        public MxCommandReply Execute(StaCommand command)
+        {
+            lock (gate)
+            {
+                correlationIds.Add(command.CorrelationId);
+            }
+
+            Started.Set();
+            release.Wait(TimeSpan.FromSeconds(5));
+
+            return new MxCommandReply
+            {
+                ProtocolStatus = new ProtocolStatus
+                {
+                    Code = ProtocolStatusCode.Ok,
+                    Message = "OK",
+                },
+            };
+        }
+
+        public void Release()
+        {
+            release.Set();
+        }
+    }
+
+    private sealed class ThrowingCommandExecutor : IStaCommandExecutor
+    {
+        private readonly Exception exception;
+
+        public ThrowingCommandExecutor(Exception exception)
+        {
+            this.exception = exception;
+        }
+
+        public MxCommandReply Execute(StaCommand command)
+        {
+            throw exception;
+        }
+    }
+
+    private sealed class NoopComApartmentInitializer : IStaComApartmentInitializer
+    {
+        public void Initialize()
+        {
+        }
+
+        public void Uninitialize()
+        {
+        }
+    }
+}

src/MxGateway.Worker.Tests/Sta/StaRuntimeTests.cs

@@ -0,0 +1,152 @@
+using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+using MxGateway.Worker.Sta;
+
+namespace MxGateway.Worker.Tests.Sta;
+
+public sealed class StaRuntimeTests
+{
+    [Fact]
+    public async Task InvokeAsync_ExecutesCommandOnStaThread()
+    {
+        RecordingComApartmentInitializer initializer = new();
+        using StaRuntime runtime = CreateRuntime(initializer);
+
+        runtime.Start();
+
+        StaCommandObservation observation = await runtime.InvokeAsync(
+            () => new StaCommandObservation(
+                Thread.CurrentThread.ManagedThreadId,
+                Thread.CurrentThread.GetApartmentState()));
+
+        Assert.Equal(runtime.StaThreadId, observation.ThreadId);
+        Assert.Equal(initializer.InitializeThreadId, observation.ThreadId);
+        Assert.Equal(ApartmentState.STA, observation.ApartmentState);
+    }
+
+    [Fact]
+    public async Task InvokeAsync_WakesIdlePumpForQueuedCommand()
+    {
+        RecordingComApartmentInitializer initializer = new();
+        using StaRuntime runtime = new(
+            initializer,
+            new StaMessagePump(),
+            TimeSpan.FromSeconds(30));
+        runtime.Start();
+        Stopwatch stopwatch = Stopwatch.StartNew();
+
+        int threadId = await runtime.InvokeAsync(() => Thread.CurrentThread.ManagedThreadId);
+
+        stopwatch.Stop();
+        Assert.Equal(runtime.StaThreadId, threadId);
+        Assert.True(
+            stopwatch.Elapsed < TimeSpan.FromSeconds(2),
+            $"Command took {stopwatch.Elapsed} to execute, so the command wake event did not wake the STA promptly.");
+    }
+
+    [Fact]
+    public void Shutdown_StopsThreadAndUninitializesComApartment()
+    {
+        RecordingComApartmentInitializer initializer = new();
+        using StaRuntime runtime = CreateRuntime(initializer);
+        runtime.Start();
+
+        bool stopped = runtime.Shutdown(TimeSpan.FromSeconds(2));
+
+        Assert.True(stopped);
+        Assert.False(runtime.IsRunning);
+        Assert.Equal(1, initializer.InitializeCount);
+        Assert.Equal(1, initializer.UninitializeCount);
+        Assert.Equal(initializer.InitializeThreadId, initializer.UninitializeThreadId);
+    }
+
+    [Fact]
+    public void LastActivityUtc_UpdatesWhilePumpIsIdle()
+    {
+        RecordingComApartmentInitializer initializer = new();
+        using StaRuntime runtime = CreateRuntime(initializer);
+        runtime.Start();
+        DateTimeOffset firstActivity = runtime.LastActivityUtc;
+
+        bool updated = SpinWait.SpinUntil(
+            () => runtime.LastActivityUtc > firstActivity,
+            TimeSpan.FromSeconds(2));
+
+        Assert.True(updated);
+    }
+
+    [Fact]
+    public async Task InvokeAsync_CommandException_FaultsReturnedTaskWithoutStoppingRuntime()
+    {
+        RecordingComApartmentInitializer initializer = new();
+        using StaRuntime runtime = CreateRuntime(initializer);
+        runtime.Start();
+
+        InvalidOperationException exception = await Assert.ThrowsAsync<InvalidOperationException>(
+            () => runtime.InvokeAsync<int>(() => throw new InvalidOperationException("command failed")));
+
+        int threadId = await runtime.InvokeAsync(() => Thread.CurrentThread.ManagedThreadId);
+        Assert.Equal("command failed", exception.Message);
+        Assert.Equal(runtime.StaThreadId, threadId);
+    }
+
+    [Fact]
+    public async Task InvokeAsync_AfterShutdown_ReturnsFaultedTask()
+    {
+        RecordingComApartmentInitializer initializer = new();
+        using StaRuntime runtime = CreateRuntime(initializer);
+        runtime.Start();
+        runtime.Shutdown(TimeSpan.FromSeconds(2));
+
+        InvalidOperationException exception = await Assert.ThrowsAsync<InvalidOperationException>(
+            () => runtime.InvokeAsync(() => Thread.CurrentThread.ManagedThreadId));
+
+        Assert.Contains("shutting down", exception.Message);
+    }
+
+    private static StaRuntime CreateRuntime(RecordingComApartmentInitializer initializer)
+    {
+        return new StaRuntime(
+            initializer,
+            new StaMessagePump(),
+            TimeSpan.FromMilliseconds(25));
+    }
+
+    private sealed class StaCommandObservation
+    {
+        public StaCommandObservation(int threadId, ApartmentState apartmentState)
+        {
+            ThreadId = threadId;
+            ApartmentState = apartmentState;
+        }
+
+        public int ThreadId { get; }
+
+        public ApartmentState ApartmentState { get; }
+    }
+
+    private sealed class RecordingComApartmentInitializer : IStaComApartmentInitializer
+    {
+        public int InitializeCount { get; private set; }
+
+        public int UninitializeCount { get; private set; }
+
+        public int? InitializeThreadId { get; private set; }
+
+        public int? UninitializeThreadId { get; private set; }
+
+        public void Initialize()
+        {
+            InitializeCount++;
+            InitializeThreadId = Thread.CurrentThread.ManagedThreadId;
+        }
+
+        public void Uninitialize()
+        {
+            UninitializeCount++;
+            UninitializeThreadId = Thread.CurrentThread.ManagedThreadId;
+        }
+    }
+}

src/MxGateway.Worker/Conversion/HResultConversion.cs

@@ -0,0 +1,22 @@
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Worker.Conversion;
+
+public sealed class HResultConversion
+{
+    public HResultConversion(
+        int hresult,
+        ProtocolStatus protocolStatus,
+        string diagnosticMessage)
+    {
+        HResult = hresult;
+        ProtocolStatus = protocolStatus;
+        DiagnosticMessage = diagnosticMessage;
+    }
+
+    public int HResult { get; }
+
+    public ProtocolStatus ProtocolStatus { get; }
+
+    public string DiagnosticMessage { get; }
+}

src/MxGateway.Worker/Conversion/HResultConverter.cs

@@ -0,0 +1,48 @@
+using System;
+using System.Runtime.InteropServices;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Worker.Conversion;
+
+public sealed class HResultConverter
+{
+    public HResultConversion Convert(Exception exception)
+    {
+        if (exception is null)
+        {
+            throw new ArgumentNullException(nameof(exception));
+        }
+
+        int hresult = exception is COMException comException
+            ? comException.ErrorCode
+            : exception.HResult;
+
+        return new HResultConversion(
+            hresult,
+            CreateProtocolStatus(hresult, exception),
+            CreateSafeDiagnosticMessage(exception));
+    }
+
+    public ProtocolStatus CreateProtocolStatus(
+        int hresult,
+        Exception? exception = null)
+    {
+        return new ProtocolStatus
+        {
+            Code = hresult == 0 ? ProtocolStatusCode.Ok : ProtocolStatusCode.MxaccessFailure,
+            Message = exception is null
+                ? FormatHResult(hresult)
+                : $"{exception.GetType().Name}: {FormatHResult(hresult)}",
+        };
+    }
+
+    private static string CreateSafeDiagnosticMessage(Exception exception)
+    {
+        return $"{exception.GetType().FullName}: {FormatHResult(exception.HResult)}";
+    }
+
+    private static string FormatHResult(int hresult)
+    {
+        return $"HRESULT 0x{unchecked((uint)hresult):X8}";
+    }
+}

src/MxGateway.Worker/Conversion/MxStatusConversionException.cs

@@ -0,0 +1,11 @@
+using System;
+
+namespace MxGateway.Worker.Conversion;
+
+public sealed class MxStatusConversionException : Exception
+{
+    public MxStatusConversionException(string message)
+        : base(message)
+    {
+    }
+}

src/MxGateway.Worker/Conversion/MxStatusDetailText.cs

@@ -0,0 +1,54 @@
+using System.Collections.Generic;
+
+namespace MxGateway.Worker.Conversion;
+
+internal static class MxStatusDetailText
+{
+    private static readonly IReadOnlyDictionary<int, string> KnownDetails = new Dictionary<int, string>
+    {
+        [16] = "Request timed out",
+        [17] = "Platform communication error",
+        [18] = "Invalid platform ID",
+        [19] = "Invalid engine ID",
+        [20] = "Engine communication error",
+        [21] = "Invalid reference",
+        [22] = "No Galaxy Repository",
+        [23] = "Invalid object ID",
+        [24] = "Object signature mismatch",
+        [25] = "Invalid primitive ID",
+        [26] = "Invalid attribute ID",
+        [27] = "Invalid property ID",
+        [28] = "Index out of range",
+        [29] = "Data out of range",
+        [30] = "Incorrect data type",
+        [31] = "Attribute not readable",
+        [32] = "Attribute not writeable",
+        [33] = "Write access denied",
+        [34] = "Unknown error",
+        [36] = "Wrong data type",
+        [37] = "Wrong number of dimensions",
+        [38] = "Invalid index",
+        [39] = "Index out of order",
+        [40] = "Dimension does not exist",
+        [41] = "Conversion not supported",
+        [42] = "Unable to convert string",
+        [43] = "Overflow",
+        [44] = "Attribute signature mismatch",
+        [47] = "Nmx version mismatch",
+        [48] = "Nmx command not valid",
+        [49] = "Lmx version mismatch",
+        [50] = "Lmx command not valid",
+        [56] = "Secured Write",
+        [57] = "Verified Write",
+        [60] = "User did not have the necessary permissions to write",
+        [61] = "Verifier did not have the necessary permissions to verify",
+        [541] = "Conversion to intended data type is not supported",
+        [542] = "Unable to convert the input string to intended data type",
+        [8017] = "Object must be offscan to modify attributes that have an MxSecurityConfigure security classification",
+    };
+
+    public static string Lookup(int detail)
+    {
+        return KnownDetails.TryGetValue(detail, out string text) ? text : string.Empty;
+    }
+}

src/MxGateway.Worker/Conversion/MxStatusProxyConverter.cs

@@ -0,0 +1,127 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Reflection;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Worker.Conversion;
+
+public sealed class MxStatusProxyConverter
+{
+    public MxStatusProxy Convert(object status)
+    {
+        if (status is null)
+        {
+            throw new ArgumentNullException(nameof(status));
+        }
+
+        Type statusType = status.GetType();
+        int success = ReadInt32Field(status, statusType, "success");
+        int rawCategory = ReadInt32Field(status, statusType, "category");
+        int rawDetectedBy = ReadInt32Field(status, statusType, "detectedBy");
+        int detail = ReadInt32Field(status, statusType, "detail");
+
+        return new MxStatusProxy
+        {
+            Success = success,
+            Category = MapCategory(rawCategory),
+            DetectedBy = MapSource(rawDetectedBy),
+            Detail = detail,
+            RawCategory = rawCategory,
+            RawDetectedBy = rawDetectedBy,
+            DiagnosticText = MxStatusDetailText.Lookup(detail),
+        };
+    }
+
+    public IReadOnlyList<MxStatusProxy> ConvertMany(Array? statuses)
+    {
+        if (statuses is null)
+        {
+            return Array.Empty<MxStatusProxy>();
+        }
+
+        List<MxStatusProxy> converted = new(statuses.Length);
+        foreach (object? status in statuses)
+        {
+            if (status is null)
+            {
+                converted.Add(new MxStatusProxy
+                {
+                    Category = MxStatusCategory.Unknown,
+                    DetectedBy = MxStatusSource.Unknown,
+                    DiagnosticText = "Null MXSTATUS_PROXY entry.",
+                });
+                continue;
+            }
+
+            converted.Add(Convert(status));
+        }
+
+        return converted;
+    }
+
+    public string PreserveCompletionOnlyStatusBytes(byte[] statusBytes)
+    {
+        if (statusBytes is null)
+        {
+            throw new ArgumentNullException(nameof(statusBytes));
+        }
+
+        return $"completion_only_status_hex={BitConverter.ToString(statusBytes).Replace("-", string.Empty)}";
+    }
+
+    private static int ReadInt32Field(
+        object value,
+        Type valueType,
+        string fieldName)
+    {
+        FieldInfo? field = valueType.GetField(fieldName, BindingFlags.Instance | BindingFlags.Public);
+        if (field is null)
+        {
+            throw new MxStatusConversionException(
+                $"Status object type '{valueType.FullName}' does not expose required field '{fieldName}'.");
+        }
+
+        object? fieldValue = field.GetValue(value);
+        if (fieldValue is null)
+        {
+            throw new MxStatusConversionException(
+                $"Status object field '{fieldName}' on type '{valueType.FullName}' is null.");
+        }
+
+        return System.Convert.ToInt32(fieldValue, CultureInfo.InvariantCulture);
+    }
+
+    private static MxStatusCategory MapCategory(int rawCategory)
+    {
+        return rawCategory switch
+        {
+            -1 => MxStatusCategory.Unknown,
+            0 => MxStatusCategory.Ok,
+            1 => MxStatusCategory.Pending,
+            2 => MxStatusCategory.Warning,
+            3 => MxStatusCategory.CommunicationError,
+            4 => MxStatusCategory.ConfigurationError,
+            5 => MxStatusCategory.OperationalError,
+            6 => MxStatusCategory.SecurityError,
+            7 => MxStatusCategory.SoftwareError,
+            8 => MxStatusCategory.OtherError,
+            _ => MxStatusCategory.Unknown,
+        };
+    }
+
+    private static MxStatusSource MapSource(int rawDetectedBy)
+    {
+        return rawDetectedBy switch
+        {
+            -1 => MxStatusSource.Unknown,
+            0 => MxStatusSource.RequestingLmx,
+            1 => MxStatusSource.RespondingLmx,
+            2 => MxStatusSource.RequestingNmx,
+            3 => MxStatusSource.RespondingNmx,
+            4 => MxStatusSource.RequestingAutomationObject,
+            5 => MxStatusSource.RespondingAutomationObject,
+            _ => MxStatusSource.Unknown,
+        };
+    }
+}

src/MxGateway.Worker/Conversion/VariantConverter.cs

@@ -0,0 +1,522 @@
+using System;
+using System.Globalization;
+using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Worker.Conversion;
+
+public sealed class VariantConverter
+{
+    public MxValue Convert(object? value)
+    {
+        return Convert(value, MxDataType.Unspecified);
+    }
+
+    public MxValue Convert(
+        object? value,
+        MxDataType expectedDataType)
+    {
+        if (value is null || value is DBNull)
+        {
+            return CreateNullValue(value, expectedDataType);
+        }
+
+        if (value is Array array)
+        {
+            return new MxValue
+            {
+                DataType = MxDataType.Unspecified,
+                VariantType = CreateArrayVariantType(array),
+                ArrayValue = ConvertArray(array, expectedDataType),
+            };
+        }
+
+        return ConvertScalar(value, expectedDataType);
+    }
+
+    public MxArray ConvertArray(
+        Array array,
+        MxDataType expectedElementDataType = MxDataType.Unspecified)
+    {
+        if (array is null)
+        {
+            throw new ArgumentNullException(nameof(array));
+        }
+
+        MxArray mxArray = new()
+        {
+            VariantType = CreateArrayVariantType(array),
+        };
+
+        for (int dimension = 0; dimension < array.Rank; dimension++)
+        {
+            mxArray.Dimensions.Add((uint)array.GetLength(dimension));
+        }
+
+        System.Type? elementType = array.GetType().GetElementType();
+        MxDataType elementDataType = ResolveArrayElementDataType(elementType, expectedElementDataType);
+        mxArray.ElementDataType = elementDataType;
+
+        switch (elementDataType)
+        {
+            case MxDataType.Boolean:
+                mxArray.BoolValues = ConvertBoolArray(array);
+                return mxArray;
+
+            case MxDataType.Integer:
+                if (elementType == typeof(long) || elementType == typeof(ulong))
+                {
+                    mxArray.Int64Values = ConvertInt64Array(array);
+                }
+                else
+                {
+                    mxArray.Int32Values = ConvertInt32Array(array);
+                }
+
+                return mxArray;
+
+            case MxDataType.Float:
+                mxArray.FloatValues = ConvertFloatArray(array);
+                return mxArray;
+
+            case MxDataType.Double:
+                mxArray.DoubleValues = ConvertDoubleArray(array);
+                return mxArray;
+
+            case MxDataType.String:
+                mxArray.StringValues = ConvertStringArray(array);
+                return mxArray;
+
+            case MxDataType.Time:
+                mxArray.TimestampValues = ConvertTimestampArray(array);
+                return mxArray;
+
+            default:
+                mxArray.ElementDataType = MxDataType.Unknown;
+                mxArray.RawElementDataType = (int)expectedElementDataType;
+                mxArray.RawDiagnostic = CreateRawDiagnostic(array);
+                mxArray.RawValues = ConvertRawArray(array);
+                return mxArray;
+        }
+    }
+
+    private static MxValue ConvertScalar(
+        object value,
+        MxDataType expectedDataType)
+    {
+        System.Type valueType = value.GetType();
+        string variantType = GetVariantTypeName(valueType);
+
+        switch (System.Type.GetTypeCode(valueType))
+        {
+            case TypeCode.Boolean:
+                return new MxValue
+                {
+                    DataType = MxDataType.Boolean,
+                    VariantType = variantType,
+                    BoolValue = (bool)value,
+                };
+
+            case TypeCode.Byte:
+            case TypeCode.SByte:
+            case TypeCode.Int16:
+            case TypeCode.UInt16:
+            case TypeCode.Int32:
+                return new MxValue
+                {
+                    DataType = MxDataType.Integer,
+                    VariantType = variantType,
+                    Int32Value = System.Convert.ToInt32(value, CultureInfo.InvariantCulture),
+                };
+
+            case TypeCode.UInt32:
+            case TypeCode.Int64:
+                return ConvertInt64Scalar(value, variantType, expectedDataType);
+
+            case TypeCode.UInt64:
+                return ConvertUInt64Scalar((ulong)value, variantType, expectedDataType);
+
+            case TypeCode.Single:
+                return new MxValue
+                {
+                    DataType = MxDataType.Float,
+                    VariantType = variantType,
+                    FloatValue = (float)value,
+                };
+
+            case TypeCode.Double:
+                return new MxValue
+                {
+                    DataType = MxDataType.Double,
+                    VariantType = variantType,
+                    DoubleValue = (double)value,
+                };
+
+            case TypeCode.Decimal:
+                return new MxValue
+                {
+                    DataType = MxDataType.Double,
+                    VariantType = variantType,
+                    DoubleValue = System.Convert.ToDouble(value, CultureInfo.InvariantCulture),
+                    RawDiagnostic = "Decimal value projected to double.",
+                };
+
+            case TypeCode.String:
+            case TypeCode.Char:
+                return new MxValue
+                {
+                    DataType = MxDataType.String,
+                    VariantType = variantType,
+                    StringValue = System.Convert.ToString(value, CultureInfo.InvariantCulture) ?? string.Empty,
+                };
+
+            case TypeCode.DateTime:
+                return new MxValue
+                {
+                    DataType = MxDataType.Time,
+                    VariantType = variantType,
+                    TimestampValue = ToTimestamp((DateTime)value),
+                };
+
+            default:
+                return CreateRawValue(value, expectedDataType);
+        }
+    }
+
+    private static MxValue ConvertInt64Scalar(
+        object value,
+        string variantType,
+        MxDataType expectedDataType)
+    {
+        long longValue = System.Convert.ToInt64(value, CultureInfo.InvariantCulture);
+        if (expectedDataType == MxDataType.Time)
+        {
+            return new MxValue
+            {
+                DataType = MxDataType.Time,
+                VariantType = variantType,
+                TimestampValue = Timestamp.FromDateTime(DateTime.FromFileTimeUtc(longValue)),
+            };
+        }
+
+        return new MxValue
+        {
+            DataType = MxDataType.Integer,
+            VariantType = variantType,
+            Int64Value = longValue,
+        };
+    }
+
+    private static MxValue ConvertUInt64Scalar(
+        ulong value,
+        string variantType,
+        MxDataType expectedDataType)
+    {
+        if (expectedDataType == MxDataType.Time && value <= long.MaxValue)
+        {
+            return new MxValue
+            {
+                DataType = MxDataType.Time,
+                VariantType = variantType,
+                TimestampValue = Timestamp.FromDateTime(DateTime.FromFileTimeUtc((long)value)),
+            };
+        }
+
+        if (value <= long.MaxValue)
+        {
+            return new MxValue
+            {
+                DataType = MxDataType.Integer,
+                VariantType = variantType,
+                Int64Value = (long)value,
+            };
+        }
+
+        return CreateRawValue(value, expectedDataType, "UInt64 value exceeds Int64 range.");
+    }
+
+    private static MxValue CreateNullValue(
+        object? value,
+        MxDataType expectedDataType)
+    {
+        return new MxValue
+        {
+            DataType = expectedDataType == MxDataType.Unspecified ? MxDataType.NoData : expectedDataType,
+            VariantType = value is DBNull ? "VT_NULL" : "VT_EMPTY",
+            IsNull = true,
+        };
+    }
+
+    private static MxValue CreateRawValue(
+        object value,
+        MxDataType expectedDataType,
+        string? diagnosticPrefix = null)
+    {
+        string diagnostic = CreateRawDiagnostic(value);
+        if (!string.IsNullOrWhiteSpace(diagnosticPrefix))
+        {
+            diagnostic = $"{diagnosticPrefix} {diagnostic}";
+        }
+
+        return new MxValue
+        {
+            DataType = MxDataType.Unknown,
+            VariantType = GetVariantTypeName(value.GetType()),
+            RawDataType = (int)expectedDataType,
+            RawDiagnostic = diagnostic,
+            RawValue = ByteString.CopyFromUtf8(System.Convert.ToString(value, CultureInfo.InvariantCulture) ?? string.Empty),
+        };
+    }
+
+    private static BoolArray ConvertBoolArray(Array array)
+    {
+        BoolArray values = new();
+        foreach (object? item in array)
+        {
+            values.Values.Add(item is not null && System.Convert.ToBoolean(item, CultureInfo.InvariantCulture));
+        }
+
+        return values;
+    }
+
+    private static Int32Array ConvertInt32Array(Array array)
+    {
+        Int32Array values = new();
+        foreach (object? item in array)
+        {
+            values.Values.Add(item is null ? 0 : System.Convert.ToInt32(item, CultureInfo.InvariantCulture));
+        }
+
+        return values;
+    }
+
+    private static Int64Array ConvertInt64Array(Array array)
+    {
+        Int64Array values = new();
+        foreach (object? item in array)
+        {
+            values.Values.Add(item is null ? 0 : System.Convert.ToInt64(item, CultureInfo.InvariantCulture));
+        }
+
+        return values;
+    }
+
+    private static FloatArray ConvertFloatArray(Array array)
+    {
+        FloatArray values = new();
+        foreach (object? item in array)
+        {
+            values.Values.Add(item is null ? 0 : System.Convert.ToSingle(item, CultureInfo.InvariantCulture));
+        }
+
+        return values;
+    }
+
+    private static DoubleArray ConvertDoubleArray(Array array)
+    {
+        DoubleArray values = new();
+        foreach (object? item in array)
+        {
+            values.Values.Add(item is null ? 0 : System.Convert.ToDouble(item, CultureInfo.InvariantCulture));
+        }
+
+        return values;
+    }
+
+    private static StringArray ConvertStringArray(Array array)
+    {
+        StringArray values = new();
+        foreach (object? item in array)
+        {
+            values.Values.Add(item is null ? string.Empty : System.Convert.ToString(item, CultureInfo.InvariantCulture) ?? string.Empty);
+        }
+
+        return values;
+    }
+
+    private static TimestampArray ConvertTimestampArray(Array array)
+    {
+        TimestampArray values = new();
+        foreach (object? item in array)
+        {
+            if (item is null)
+            {
+                values.Values.Add(Timestamp.FromDateTime(new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)));
+            }
+            else if (item is DateTime dateTime)
+            {
+                values.Values.Add(ToTimestamp(dateTime));
+            }
+            else
+            {
+                long fileTime = System.Convert.ToInt64(item, CultureInfo.InvariantCulture);
+                values.Values.Add(Timestamp.FromDateTime(DateTime.FromFileTimeUtc(fileTime)));
+            }
+        }
+
+        return values;
+    }
+
+    private static RawArray ConvertRawArray(Array array)
+    {
+        RawArray values = new();
+        foreach (object? item in array)
+        {
+            string rawValue = item is null
+                ? string.Empty
+                : System.Convert.ToString(item, CultureInfo.InvariantCulture) ?? string.Empty;
+            values.Values.Add(ByteString.CopyFromUtf8(rawValue));
+        }
+
+        return values;
+    }
+
+    private static MxDataType ResolveArrayElementDataType(
+        System.Type? elementType,
+        MxDataType expectedElementDataType)
+    {
+        if (expectedElementDataType != MxDataType.Unspecified)
+        {
+            return expectedElementDataType;
+        }
+
+        if (elementType == typeof(bool))
+        {
+            return MxDataType.Boolean;
+        }
+
+        if (elementType == typeof(byte)
+            || elementType == typeof(sbyte)
+            || elementType == typeof(short)
+            || elementType == typeof(ushort)
+            || elementType == typeof(int)
+            || elementType == typeof(uint)
+            || elementType == typeof(long)
+            || elementType == typeof(ulong))
+        {
+            return MxDataType.Integer;
+        }
+
+        if (elementType == typeof(float))
+        {
+            return MxDataType.Float;
+        }
+
+        if (elementType == typeof(double) || elementType == typeof(decimal))
+        {
+            return MxDataType.Double;
+        }
+
+        if (elementType == typeof(string) || elementType == typeof(char))
+        {
+            return MxDataType.String;
+        }
+
+        if (elementType == typeof(DateTime))
+        {
+            return MxDataType.Time;
+        }
+
+        return MxDataType.Unknown;
+    }
+
+    private static Timestamp ToTimestamp(DateTime dateTime)
+    {
+        DateTime utcDateTime = dateTime.Kind switch
+        {
+            DateTimeKind.Utc => dateTime,
+            DateTimeKind.Local => dateTime.ToUniversalTime(),
+            _ => DateTime.SpecifyKind(dateTime, DateTimeKind.Utc),
+        };
+
+        return Timestamp.FromDateTime(utcDateTime);
+    }
+
+    private static string CreateArrayVariantType(Array array)
+    {
+        System.Type? elementType = array.GetType().GetElementType();
+        return $"SAFEARRAY({GetVariantTypeName(elementType)})";
+    }
+
+    private static string GetVariantTypeName(System.Type? type)
+    {
+        if (type is null)
+        {
+            return "VT_EMPTY";
+        }
+
+        System.Type nonNullableType = Nullable.GetUnderlyingType(type) ?? type;
+        if (nonNullableType == typeof(bool))
+        {
+            return "VT_BOOL";
+        }
+
+        if (nonNullableType == typeof(byte))
+        {
+            return "VT_UI1";
+        }
+
+        if (nonNullableType == typeof(sbyte))
+        {
+            return "VT_I1";
+        }
+
+        if (nonNullableType == typeof(short))
+        {
+            return "VT_I2";
+        }
+
+        if (nonNullableType == typeof(ushort))
+        {
+            return "VT_UI2";
+        }
+
+        if (nonNullableType == typeof(int))
+        {
+            return "VT_I4";
+        }
+
+        if (nonNullableType == typeof(uint))
+        {
+            return "VT_UI4";
+        }
+
+        if (nonNullableType == typeof(long))
+        {
+            return "VT_I8";
+        }
+
+        if (nonNullableType == typeof(ulong))
+        {
+            return "VT_UI8";
+        }
+
+        if (nonNullableType == typeof(float))
+        {
+            return "VT_R4";
+        }
+
+        if (nonNullableType == typeof(double) || nonNullableType == typeof(decimal))
+        {
+            return "VT_R8";
+        }
+
+        if (nonNullableType == typeof(string) || nonNullableType == typeof(char))
+        {
+            return "VT_BSTR";
+        }
+
+        if (nonNullableType == typeof(DateTime))
+        {
+            return "VT_DATE";
+        }
+
+        return $"CLR:{nonNullableType.FullName}";
+    }
+
+    private static string CreateRawDiagnostic(object value)
+    {
+        return $"Unsupported variant projection for CLR type '{value.GetType().FullName}'.";
+    }
+}

src/MxGateway.Worker/Ipc/WorkerPipeSession.cs

@@ -15,6 +15,7 @@ public sealed class WorkerPipeSession
     private readonly Func<int> _processIdProvider;
     private readonly WorkerFrameReader _reader;
     private readonly WorkerFrameWriter _writer;
+    private MxAccessStaSession? _mxAccessStaSession;
     private long _nextSequence;
 
     public WorkerPipeSession(
@@ -42,7 +43,7 @@ public sealed class WorkerPipeSession
 
     public Task CompleteStartupHandshakeAsync(CancellationToken cancellationToken = default)
     {
-        return CompleteStartupHandshakeAsync(_ => Task.CompletedTask, cancellationToken);
+        return CompleteStartupHandshakeAsync(InitializeMxAccessAsync, cancellationToken);
     }
 
     public async Task CompleteStartupHandshakeAsync(
@@ -54,20 +55,44 @@ public sealed class WorkerPipeSession
             throw new ArgumentNullException(nameof(initializeMxAccessAsync));
         }
 
+        await CompleteStartupHandshakeAsync(
+            async innerCancellationToken =>
+            {
+                await initializeMxAccessAsync(innerCancellationToken).ConfigureAwait(false);
+                return CreateWorkerReady();
+            },
+            cancellationToken).ConfigureAwait(false);
+    }
+
+    public async Task CompleteStartupHandshakeAsync(
+        Func<CancellationToken, Task<WorkerReady>> initializeMxAccessAsync,
+        CancellationToken cancellationToken = default)
+    {
+        if (initializeMxAccessAsync is null)
+        {
+            throw new ArgumentNullException(nameof(initializeMxAccessAsync));
+        }
+
         try
         {
             WorkerEnvelope envelope = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false);
             ValidateGatewayHello(envelope);
 
             await WriteWorkerHelloAsync(cancellationToken).ConfigureAwait(false);
-            await initializeMxAccessAsync(cancellationToken).ConfigureAwait(false);
-            await WriteWorkerReadyAsync(cancellationToken).ConfigureAwait(false);
+            WorkerReady ready = await initializeMxAccessAsync(cancellationToken).ConfigureAwait(false);
+            await WriteWorkerReadyAsync(ready, cancellationToken).ConfigureAwait(false);
         }
         catch (WorkerFrameProtocolException exception)
         {
             await TryWriteFaultAsync(exception, cancellationToken).ConfigureAwait(false);
             throw;
         }
+        catch (Exception exception) when (exception is not OperationCanceledException)
+        {
+            await TryWriteFaultAsync(MxAccessCreationException.From(exception), cancellationToken)
+                .ConfigureAwait(false);
+            throw;
+        }
     }
 
     private void ValidateGatewayHello(WorkerEnvelope envelope)
@@ -108,17 +133,11 @@ public sealed class WorkerPipeSession
             cancellationToken);
     }
 
-    private Task WriteWorkerReadyAsync(CancellationToken cancellationToken)
+    private Task WriteWorkerReadyAsync(
+        WorkerReady ready,
+        CancellationToken cancellationToken)
     {
-        return _writer.WriteAsync(
-            CreateEnvelope(new WorkerReady
-            {
-                WorkerProcessId = _processIdProvider(),
-                MxaccessProgid = MxAccessInteropInfo.ProgId,
-                MxaccessClsid = MxAccessInteropInfo.Clsid,
-                ReadyTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
-            }),
-            cancellationToken);
+        return _writer.WriteAsync(CreateEnvelope(ready), cancellationToken);
     }
 
     private async Task TryWriteFaultAsync(
@@ -140,6 +159,25 @@ public sealed class WorkerPipeSession
         }
     }
 
+    private async Task TryWriteFaultAsync(
+        MxAccessCreationException exception,
+        CancellationToken cancellationToken)
+    {
+        try
+        {
+            await _writer
+                .WriteAsync(CreateEnvelope(CreateFault(exception)), cancellationToken)
+                .ConfigureAwait(false);
+        }
+        catch (Exception faultWriteException) when (
+            faultWriteException is IOException
+            || faultWriteException is ObjectDisposedException
+            || faultWriteException is WorkerFrameProtocolException)
+        {
+            // The MXAccess creation failure is the actionable error.
+        }
+    }
+
     private WorkerEnvelope CreateEnvelope(WorkerHello hello)
     {
         return CreateBaseEnvelope(hello);
@@ -191,6 +229,34 @@ public sealed class WorkerPipeSession
         return unchecked((ulong)Interlocked.Increment(ref _nextSequence));
     }
 
+    private async Task<WorkerReady> InitializeMxAccessAsync(CancellationToken cancellationToken)
+    {
+        _mxAccessStaSession = new MxAccessStaSession();
+        try
+        {
+            return await _mxAccessStaSession
+                .StartAsync(_processIdProvider(), cancellationToken)
+                .ConfigureAwait(false);
+        }
+        catch
+        {
+            _mxAccessStaSession.Dispose();
+            _mxAccessStaSession = null;
+            throw;
+        }
+    }
+
+    private WorkerReady CreateWorkerReady()
+    {
+        return new WorkerReady
+        {
+            WorkerProcessId = _processIdProvider(),
+            MxaccessProgid = MxAccessInteropInfo.ProgId,
+            MxaccessClsid = MxAccessInteropInfo.Clsid,
+            ReadyTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+        };
+    }
+
     private static WorkerFault CreateFault(WorkerFrameProtocolException exception)
     {
         return new WorkerFault
@@ -206,6 +272,29 @@ public sealed class WorkerPipeSession
         };
     }
 
+    private static WorkerFault CreateFault(MxAccessCreationException exception)
+    {
+        WorkerFault fault = new()
+        {
+            Category = WorkerFaultCategory.MxaccessCreationFailed,
+            ExceptionType = exception.InnerException?.GetType().FullName ?? exception.GetType().FullName ?? string.Empty,
+            DiagnosticMessage = exception.Message,
+            ProtocolStatus = new ProtocolStatus
+            {
+                Code = ProtocolStatusCode.WorkerUnavailable,
+                Message = exception.Message,
+            },
+        };
+
+        int? hresult = MxAccessCreationException.ExtractHResult(exception);
+        if (hresult.HasValue)
+        {
+            fault.Hresult = hresult.Value;
+        }
+
+        return fault;
+    }
+
     private static WorkerFaultCategory MapFaultCategory(WorkerFrameProtocolErrorCode errorCode)
     {
         return errorCode switch

src/MxGateway.Worker/MxAccess/IMxAccessComObjectFactory.cs

@@ -0,0 +1,6 @@
+namespace MxGateway.Worker.MxAccess;
+
+public interface IMxAccessComObjectFactory
+{
+    object Create();
+}

src/MxGateway.Worker/MxAccess/IMxAccessEventSink.cs

@@ -0,0 +1,8 @@
+namespace MxGateway.Worker.MxAccess;
+
+public interface IMxAccessEventSink
+{
+    void Attach(object mxAccessComObject);
+
+    void Detach();
+}

src/MxGateway.Worker/MxAccess/IMxAccessServer.cs

@@ -0,0 +1,8 @@
+namespace MxGateway.Worker.MxAccess;
+
+public interface IMxAccessServer
+{
+    int Register(string clientName);
+
+    void Unregister(int serverHandle);
+}

src/MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs

@@ -0,0 +1,66 @@
+using ArchestrA.MxAccess;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessBaseEventSink : IMxAccessEventSink
+{
+    private LMXProxyServerClass? server;
+
+    public void Attach(object mxAccessComObject)
+    {
+        server = (LMXProxyServerClass)mxAccessComObject;
+        server.OnDataChange += OnDataChange;
+        server.OnWriteComplete += OnWriteComplete;
+        server.OperationComplete += OperationComplete;
+        server.OnBufferedDataChange += OnBufferedDataChange;
+    }
+
+    public void Detach()
+    {
+        if (server is null)
+        {
+            return;
+        }
+
+        server.OnDataChange -= OnDataChange;
+        server.OnWriteComplete -= OnWriteComplete;
+        server.OperationComplete -= OperationComplete;
+        server.OnBufferedDataChange -= OnBufferedDataChange;
+        server = null;
+    }
+
+    private static void OnDataChange(
+        int hLMXServerHandle,
+        int phItemHandle,
+        object pvItemValue,
+        int pwItemQuality,
+        object pftItemTimeStamp,
+        ref MXSTATUS_PROXY[] pVars)
+    {
+    }
+
+    private static void OnWriteComplete(
+        int hLMXServerHandle,
+        int phItemHandle,
+        ref MXSTATUS_PROXY[] pVars)
+    {
+    }
+
+    private static void OperationComplete(
+        int hLMXServerHandle,
+        int phItemHandle,
+        ref MXSTATUS_PROXY[] pVars)
+    {
+    }
+
+    private static void OnBufferedDataChange(
+        int hLMXServerHandle,
+        int phItemHandle,
+        MxDataType dtDataType,
+        object pvItemValue,
+        object pwItemQuality,
+        object pftItemTimeStamp,
+        ref MXSTATUS_PROXY[] pVars)
+    {
+    }
+}

src/MxGateway.Worker/MxAccess/MxAccessComObjectFactory.cs

@@ -0,0 +1,11 @@
+using ArchestrA.MxAccess;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessComObjectFactory : IMxAccessComObjectFactory
+{
+    public object Create()
+    {
+        return new LMXProxyServerClass();
+    }
+}

src/MxGateway.Worker/MxAccess/MxAccessComServer.cs

@@ -0,0 +1,59 @@
+using System;
+using System.Reflection;
+using System.Runtime.ExceptionServices;
+using ArchestrA.MxAccess;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessComServer : IMxAccessServer
+{
+    private readonly object mxAccessComObject;
+
+    public MxAccessComServer(object mxAccessComObject)
+    {
+        this.mxAccessComObject = mxAccessComObject ?? throw new ArgumentNullException(nameof(mxAccessComObject));
+    }
+
+    public int Register(string clientName)
+    {
+        if (mxAccessComObject is ILMXProxyServer mxAccessServer)
+        {
+            return mxAccessServer.Register(clientName);
+        }
+
+        return (int)Invoke(nameof(Register), clientName);
+    }
+
+    public void Unregister(int serverHandle)
+    {
+        if (mxAccessComObject is ILMXProxyServer mxAccessServer)
+        {
+            mxAccessServer.Unregister(serverHandle);
+            return;
+        }
+
+        Invoke(nameof(Unregister), serverHandle);
+    }
+
+    private object Invoke(
+        string methodName,
+        params object[] arguments)
+    {
+        try
+        {
+            return mxAccessComObject
+                .GetType()
+                .InvokeMember(
+                    methodName,
+                    BindingFlags.Instance | BindingFlags.Public | BindingFlags.InvokeMethod,
+                    binder: null,
+                    target: mxAccessComObject,
+                    args: arguments);
+        }
+        catch (TargetInvocationException exception) when (exception.InnerException is not null)
+        {
+            ExceptionDispatchInfo.Capture(exception.InnerException).Throw();
+            throw;
+        }
+    }
+}

src/MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs

@@ -0,0 +1,103 @@
+using System;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.Conversion;
+using MxGateway.Worker.Sta;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessCommandExecutor : IStaCommandExecutor
+{
+    private readonly MxAccessSession session;
+    private readonly VariantConverter variantConverter;
+
+    public MxAccessCommandExecutor(MxAccessSession session)
+        : this(session, new VariantConverter())
+    {
+    }
+
+    public MxAccessCommandExecutor(
+        MxAccessSession session,
+        VariantConverter variantConverter)
+    {
+        this.session = session ?? throw new ArgumentNullException(nameof(session));
+        this.variantConverter = variantConverter ?? throw new ArgumentNullException(nameof(variantConverter));
+    }
+
+    public MxCommandReply Execute(StaCommand command)
+    {
+        if (command is null)
+        {
+            throw new ArgumentNullException(nameof(command));
+        }
+
+        return command.Kind switch
+        {
+            MxCommandKind.Register => ExecuteRegister(command),
+            MxCommandKind.Unregister => ExecuteUnregister(command),
+            _ => CreateInvalidRequestReply(command, $"Unsupported MXAccess command kind {command.Kind}."),
+        };
+    }
+
+    private MxCommandReply ExecuteRegister(StaCommand command)
+    {
+        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Register)
+        {
+            return CreateInvalidRequestReply(command, "Register command payload is required.");
+        }
+
+        int serverHandle = session.Register(command.Command.Register.ClientName);
+        MxCommandReply reply = CreateOkReply(command);
+        reply.ReturnValue = variantConverter.Convert(serverHandle);
+        reply.Register = new RegisterReply
+        {
+            ServerHandle = serverHandle,
+        };
+
+        return reply;
+    }
+
+    private MxCommandReply ExecuteUnregister(StaCommand command)
+    {
+        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Unregister)
+        {
+            return CreateInvalidRequestReply(command, "Unregister command payload is required.");
+        }
+
+        session.Unregister(command.Command.Unregister.ServerHandle);
+        return CreateOkReply(command);
+    }
+
+    private static MxCommandReply CreateOkReply(StaCommand command)
+    {
+        return new MxCommandReply
+        {
+            SessionId = command.SessionId,
+            CorrelationId = command.CorrelationId,
+            Kind = command.Kind,
+            Hresult = 0,
+            ProtocolStatus = new ProtocolStatus
+            {
+                Code = ProtocolStatusCode.Ok,
+                Message = "OK",
+            },
+        };
+    }
+
+    private static MxCommandReply CreateInvalidRequestReply(
+        StaCommand command,
+        string message)
+    {
+        return new MxCommandReply
+        {
+            SessionId = command.SessionId,
+            CorrelationId = command.CorrelationId,
+            Kind = command.Kind,
+            ProtocolStatus = new ProtocolStatus
+            {
+                Code = ProtocolStatusCode.InvalidRequest,
+                Message = message,
+            },
+            DiagnosticMessage = message,
+        };
+    }
+}

src/MxGateway.Worker/MxAccess/MxAccessCreationException.cs

@@ -0,0 +1,48 @@
+using System;
+using System.Runtime.InteropServices;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessCreationException : Exception
+{
+    public MxAccessCreationException(Exception innerException)
+        : base(
+            $"Failed to create MXAccess COM object {MxAccessInteropInfo.ComClassName} ({MxAccessInteropInfo.ProgId}).",
+            innerException)
+    {
+        AttemptedProgId = MxAccessInteropInfo.ProgId;
+        AttemptedClsid = MxAccessInteropInfo.Clsid;
+        AttemptedComClassName = MxAccessInteropInfo.ComClassName;
+        HResult = innerException.HResult;
+    }
+
+    public string AttemptedProgId { get; }
+
+    public string AttemptedClsid { get; }
+
+    public string AttemptedComClassName { get; }
+
+    public int? CapturedHResult => HResult == 0 ? null : HResult;
+
+    public static MxAccessCreationException From(Exception exception)
+    {
+        return exception is MxAccessCreationException creationException
+            ? creationException
+            : new MxAccessCreationException(exception);
+    }
+
+    public static int? ExtractHResult(Exception exception)
+    {
+        if (exception is MxAccessCreationException creationException)
+        {
+            return creationException.CapturedHResult;
+        }
+
+        if (exception is COMException comException)
+        {
+            return comException.HResult;
+        }
+
+        return exception.HResult == 0 ? null : exception.HResult;
+    }
+}

src/MxGateway.Worker/MxAccess/MxAccessHandleRegistry.cs

@@ -0,0 +1,31 @@
+using System.Collections.Generic;
+using System.Linq;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessHandleRegistry
+{
+    private readonly Dictionary<int, RegisteredServerHandle> serverHandles = new();
+
+    public IReadOnlyList<RegisteredServerHandle> ServerHandles => serverHandles
+        .Values
+        .OrderBy(handle => handle.ServerHandle)
+        .ToArray();
+
+    public void RegisterServerHandle(
+        int serverHandle,
+        string clientName)
+    {
+        serverHandles[serverHandle] = new RegisteredServerHandle(serverHandle, clientName);
+    }
+
+    public void UnregisterServerHandle(int serverHandle)
+    {
+        serverHandles.Remove(serverHandle);
+    }
+
+    public bool ContainsServerHandle(int serverHandle)
+    {
+        return serverHandles.ContainsKey(serverHandle);
+    }
+}

src/MxGateway.Worker/MxAccess/MxAccessSession.cs

@@ -0,0 +1,133 @@
+using System;
+using System.Runtime.InteropServices;
+using Google.Protobuf.WellKnownTypes;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessSession : IDisposable
+{
+    private readonly object mxAccessComObject;
+    private readonly IMxAccessServer mxAccessServer;
+    private readonly IMxAccessEventSink eventSink;
+    private readonly MxAccessHandleRegistry handleRegistry;
+    private bool disposed;
+
+    private MxAccessSession(
+        object mxAccessComObject,
+        IMxAccessServer mxAccessServer,
+        IMxAccessEventSink eventSink,
+        MxAccessHandleRegistry handleRegistry,
+        int creationThreadId)
+    {
+        this.mxAccessComObject = mxAccessComObject ?? throw new ArgumentNullException(nameof(mxAccessComObject));
+        this.mxAccessServer = mxAccessServer ?? throw new ArgumentNullException(nameof(mxAccessServer));
+        this.eventSink = eventSink ?? throw new ArgumentNullException(nameof(eventSink));
+        this.handleRegistry = handleRegistry ?? throw new ArgumentNullException(nameof(handleRegistry));
+        CreationThreadId = creationThreadId;
+    }
+
+    public int CreationThreadId { get; }
+
+    public MxAccessHandleRegistry HandleRegistry => handleRegistry;
+
+    public WorkerReady CreateWorkerReady(int workerProcessId)
+    {
+        return new WorkerReady
+        {
+            WorkerProcessId = workerProcessId,
+            MxaccessProgid = MxAccessInteropInfo.ProgId,
+            MxaccessClsid = MxAccessInteropInfo.Clsid,
+            ReadyTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+        };
+    }
+
+    public static MxAccessSession Create(
+        IMxAccessComObjectFactory factory,
+        IMxAccessEventSink eventSink)
+    {
+        if (factory is null)
+        {
+            throw new ArgumentNullException(nameof(factory));
+        }
+
+        if (eventSink is null)
+        {
+            throw new ArgumentNullException(nameof(eventSink));
+        }
+
+        object? mxAccessComObject = null;
+
+        try
+        {
+            mxAccessComObject = factory.Create();
+            if (mxAccessComObject is null)
+            {
+                throw new InvalidOperationException("MXAccess COM factory returned null.");
+            }
+
+            eventSink.Attach(mxAccessComObject);
+
+            return new MxAccessSession(
+                mxAccessComObject,
+                new MxAccessComServer(mxAccessComObject),
+                eventSink,
+                new MxAccessHandleRegistry(),
+                Environment.CurrentManagedThreadId);
+        }
+        catch (Exception exception)
+        {
+            eventSink.Detach();
+
+            if (mxAccessComObject is not null && Marshal.IsComObject(mxAccessComObject))
+            {
+                Marshal.FinalReleaseComObject(mxAccessComObject);
+            }
+
+            throw MxAccessCreationException.From(exception);
+        }
+    }
+
+    public int Register(string clientName)
+    {
+        ThrowIfDisposed();
+
+        int serverHandle = mxAccessServer.Register(clientName);
+        handleRegistry.RegisterServerHandle(serverHandle, clientName);
+
+        return serverHandle;
+    }
+
+    public void Unregister(int serverHandle)
+    {
+        ThrowIfDisposed();
+
+        mxAccessServer.Unregister(serverHandle);
+        handleRegistry.UnregisterServerHandle(serverHandle);
+    }
+
+    public void Dispose()
+    {
+        if (disposed)
+        {
+            return;
+        }
+
+        eventSink.Detach();
+
+        if (Marshal.IsComObject(mxAccessComObject))
+        {
+            Marshal.FinalReleaseComObject(mxAccessComObject);
+        }
+
+        disposed = true;
+    }
+
+    private void ThrowIfDisposed()
+    {
+        if (disposed)
+        {
+            throw new ObjectDisposedException(nameof(MxAccessSession));
+        }
+    }
+}

src/MxGateway.Worker/MxAccess/MxAccessStaSession.cs

@@ -0,0 +1,101 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.Sta;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessStaSession : IDisposable
+{
+    private readonly IMxAccessComObjectFactory factory;
+    private readonly IMxAccessEventSink eventSink;
+    private readonly StaRuntime staRuntime;
+    private StaCommandDispatcher? commandDispatcher;
+    private MxAccessSession? session;
+    private bool disposed;
+
+    public MxAccessStaSession()
+        : this(
+            new StaRuntime(),
+            new MxAccessComObjectFactory(),
+            new MxAccessBaseEventSink())
+    {
+    }
+
+    public MxAccessStaSession(
+        StaRuntime staRuntime,
+        IMxAccessComObjectFactory factory,
+        IMxAccessEventSink eventSink)
+    {
+        this.staRuntime = staRuntime ?? throw new ArgumentNullException(nameof(staRuntime));
+        this.factory = factory ?? throw new ArgumentNullException(nameof(factory));
+        this.eventSink = eventSink ?? throw new ArgumentNullException(nameof(eventSink));
+    }
+
+    public Task<WorkerReady> StartAsync(
+        int workerProcessId,
+        CancellationToken cancellationToken = default)
+    {
+        staRuntime.Start();
+
+        return staRuntime.InvokeAsync(
+            () =>
+            {
+                if (session is not null)
+                {
+                    throw new InvalidOperationException("MXAccess COM session has already been created.");
+                }
+
+                session = MxAccessSession.Create(factory, eventSink);
+                commandDispatcher = new StaCommandDispatcher(
+                    staRuntime,
+                    new MxAccessCommandExecutor(session));
+
+                return session.CreateWorkerReady(workerProcessId);
+            },
+            cancellationToken);
+    }
+
+    public Task<MxCommandReply> DispatchAsync(StaCommand command)
+    {
+        if (commandDispatcher is null)
+        {
+            throw new InvalidOperationException("MXAccess COM session has not been started.");
+        }
+
+        return commandDispatcher.DispatchAsync(command);
+    }
+
+    public Task<IReadOnlyList<RegisteredServerHandle>> GetRegisteredServerHandlesAsync(
+        CancellationToken cancellationToken = default)
+    {
+        if (session is null)
+        {
+            throw new InvalidOperationException("MXAccess COM session has not been started.");
+        }
+
+        return staRuntime.InvokeAsync(
+            () => session.HandleRegistry.ServerHandles,
+            cancellationToken);
+    }
+
+    public void Dispose()
+    {
+        if (disposed)
+        {
+            return;
+        }
+
+        commandDispatcher?.RequestShutdown();
+
+        if (session is not null)
+        {
+            staRuntime.InvokeAsync(() => session.Dispose()).GetAwaiter().GetResult();
+        }
+
+        staRuntime.Dispose();
+        disposed = true;
+    }
+}

src/MxGateway.Worker/MxAccess/RegisteredServerHandle.cs

@@ -0,0 +1,16 @@
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class RegisteredServerHandle
+{
+    public RegisteredServerHandle(
+        int serverHandle,
+        string clientName)
+    {
+        ServerHandle = serverHandle;
+        ClientName = clientName;
+    }
+
+    public int ServerHandle { get; }
+
+    public string ClientName { get; }
+}

src/MxGateway.Worker/Sta/IStaComApartmentInitializer.cs

@@ -0,0 +1,8 @@
+namespace MxGateway.Worker.Sta;
+
+public interface IStaComApartmentInitializer
+{
+    void Initialize();
+
+    void Uninitialize();
+}

src/MxGateway.Worker/Sta/IStaCommandExecutor.cs

@@ -0,0 +1,8 @@
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Worker.Sta;
+
+public interface IStaCommandExecutor
+{
+    MxCommandReply Execute(StaCommand command);
+}

src/MxGateway.Worker/Sta/IStaWorkItem.cs

@@ -0,0 +1,8 @@
+namespace MxGateway.Worker.Sta;
+
+internal interface IStaWorkItem
+{
+    void CancelBeforeExecution();
+
+    void Execute();
+}

src/MxGateway.Worker/Sta/StaComApartmentInitializer.cs

@@ -0,0 +1,31 @@
+using System;
+using System.Runtime.InteropServices;
+
+namespace MxGateway.Worker.Sta;
+
+public sealed class StaComApartmentInitializer : IStaComApartmentInitializer
+{
+    private const uint CoInitializeApartmentThreaded = 0x2;
+    private const int SOk = 0;
+    private const int SFalse = 1;
+
+    public void Initialize()
+    {
+        int hresult = CoInitializeEx(IntPtr.Zero, CoInitializeApartmentThreaded);
+        if (hresult != SOk && hresult != SFalse)
+        {
+            throw new COMException("Failed to initialize the worker STA COM apartment.", hresult);
+        }
+    }
+
+    public void Uninitialize()
+    {
+        CoUninitialize();
+    }
+
+    [DllImport("ole32.dll")]
+    private static extern int CoInitializeEx(IntPtr reserved, uint coInit);
+
+    [DllImport("ole32.dll")]
+    private static extern void CoUninitialize();
+}

src/MxGateway.Worker/Sta/StaCommand.cs

@@ -0,0 +1,47 @@
+using System;
+using System.Threading;
+using Google.Protobuf.WellKnownTypes;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Worker.Sta;
+
+public sealed class StaCommand
+{
+    public StaCommand(
+        string sessionId,
+        string correlationId,
+        MxCommand command,
+        Timestamp? enqueueTimestamp = null,
+        CancellationToken cancellationToken = default)
+    {
+        if (string.IsNullOrWhiteSpace(sessionId))
+        {
+            throw new ArgumentException("STA command requires a session id.", nameof(sessionId));
+        }
+
+        if (string.IsNullOrWhiteSpace(correlationId))
+        {
+            throw new ArgumentException("STA command requires a correlation id.", nameof(correlationId));
+        }
+
+        SessionId = sessionId;
+        CorrelationId = correlationId;
+        Command = command ?? throw new ArgumentNullException(nameof(command));
+        EnqueueTimestamp = enqueueTimestamp ?? Timestamp.FromDateTime(DateTime.UtcNow);
+        CancellationToken = cancellationToken;
+    }
+
+    public string SessionId { get; }
+
+    public string CorrelationId { get; }
+
+    public MxCommand Command { get; }
+
+    public Timestamp EnqueueTimestamp { get; }
+
+    public CancellationToken CancellationToken { get; }
+
+    public MxCommandKind Kind => Command.Kind;
+
+    public string MethodName => Kind.ToString();
+}

src/MxGateway.Worker/Sta/StaCommandDispatcher.cs

@@ -0,0 +1,267 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.Conversion;
+
+namespace MxGateway.Worker.Sta;
+
+public sealed class StaCommandDispatcher
+{
+    private readonly HResultConverter hresultConverter;
+    private readonly IStaCommandExecutor commandExecutor;
+    private readonly Queue<QueuedStaCommand> commandQueue = new();
+    private readonly StaRuntime staRuntime;
+    private readonly object gate = new();
+    private bool drainActive;
+    private bool shutdownRequested;
+    private string currentCommandCorrelationId = string.Empty;
+
+    public StaCommandDispatcher(
+        StaRuntime staRuntime,
+        IStaCommandExecutor commandExecutor)
+        : this(staRuntime, commandExecutor, new HResultConverter())
+    {
+    }
+
+    public StaCommandDispatcher(
+        StaRuntime staRuntime,
+        IStaCommandExecutor commandExecutor,
+        HResultConverter hresultConverter)
+    {
+        this.staRuntime = staRuntime ?? throw new ArgumentNullException(nameof(staRuntime));
+        this.commandExecutor = commandExecutor ?? throw new ArgumentNullException(nameof(commandExecutor));
+        this.hresultConverter = hresultConverter ?? throw new ArgumentNullException(nameof(hresultConverter));
+    }
+
+    public int PendingCommandCount
+    {
+        get
+        {
+            lock (gate)
+            {
+                return commandQueue.Count;
+            }
+        }
+    }
+
+    public string CurrentCommandCorrelationId
+    {
+        get
+        {
+            lock (gate)
+            {
+                return currentCommandCorrelationId;
+            }
+        }
+    }
+
+    public Task<MxCommandReply> DispatchAsync(StaCommand command)
+    {
+        if (command is null)
+        {
+            throw new ArgumentNullException(nameof(command));
+        }
+
+        lock (gate)
+        {
+            if (shutdownRequested)
+            {
+                return Task.FromResult(CreateRejectedReply(
+                    command,
+                    ProtocolStatusCode.WorkerUnavailable,
+                    "The STA command dispatcher is shutting down."));
+            }
+
+            QueuedStaCommand queuedCommand = new(command);
+            commandQueue.Enqueue(queuedCommand);
+
+            if (!drainActive)
+            {
+                drainActive = true;
+                _ = DrainAsync();
+            }
+
+            return queuedCommand.Task;
+        }
+    }
+
+    public void RequestShutdown()
+    {
+        lock (gate)
+        {
+            shutdownRequested = true;
+        }
+    }
+
+    public void PopulateHeartbeat(WorkerHeartbeat heartbeat)
+    {
+        if (heartbeat is null)
+        {
+            throw new ArgumentNullException(nameof(heartbeat));
+        }
+
+        lock (gate)
+        {
+            heartbeat.PendingCommandCount = (uint)commandQueue.Count;
+            heartbeat.CurrentCommandCorrelationId = currentCommandCorrelationId;
+        }
+    }
+
+    private async Task DrainAsync()
+    {
+        while (true)
+        {
+            QueuedStaCommand queuedCommand;
+            lock (gate)
+            {
+                if (commandQueue.Count == 0)
+                {
+                    drainActive = false;
+                    return;
+                }
+
+                queuedCommand = commandQueue.Dequeue();
+            }
+
+            await ExecuteQueuedCommandAsync(queuedCommand).ConfigureAwait(false);
+        }
+    }
+
+    private async Task ExecuteQueuedCommandAsync(QueuedStaCommand queuedCommand)
+    {
+        StaCommand command = queuedCommand.Command;
+        if (command.CancellationToken.IsCancellationRequested)
+        {
+            queuedCommand.Complete(CreateRejectedReply(
+                command,
+                ProtocolStatusCode.Canceled,
+                "The STA command was canceled before execution."));
+            return;
+        }
+
+        SetCurrentCommand(command.CorrelationId);
+        try
+        {
+            MxCommandReply reply = await staRuntime
+                .InvokeAsync(() => commandExecutor.Execute(command))
+                .ConfigureAwait(false);
+
+            queuedCommand.Complete(NormalizeReply(command, reply));
+        }
+        catch (Exception exception)
+        {
+            queuedCommand.Complete(CreateExceptionReply(command, exception));
+        }
+        finally
+        {
+            SetCurrentCommand(string.Empty);
+        }
+    }
+
+    private void SetCurrentCommand(string correlationId)
+    {
+        lock (gate)
+        {
+            currentCommandCorrelationId = correlationId;
+        }
+    }
+
+    private MxCommandReply NormalizeReply(
+        StaCommand command,
+        MxCommandReply reply)
+    {
+        if (reply is null)
+        {
+            return CreateRejectedReply(
+                command,
+                ProtocolStatusCode.ProtocolViolation,
+                "STA command executor returned null.");
+        }
+
+        if (string.IsNullOrWhiteSpace(reply.SessionId))
+        {
+            reply.SessionId = command.SessionId;
+        }
+
+        if (string.IsNullOrWhiteSpace(reply.CorrelationId))
+        {
+            reply.CorrelationId = command.CorrelationId;
+        }
+
+        if (reply.Kind == MxCommandKind.Unspecified)
+        {
+            reply.Kind = command.Kind;
+        }
+
+        if (reply.ProtocolStatus is null)
+        {
+            reply.ProtocolStatus = new ProtocolStatus
+            {
+                Code = ProtocolStatusCode.Ok,
+                Message = "OK",
+            };
+        }
+
+        return reply;
+    }
+
+    private MxCommandReply CreateExceptionReply(
+        StaCommand command,
+        Exception exception)
+    {
+        HResultConversion conversion = hresultConverter.Convert(exception);
+        MxCommandReply reply = CreateBaseReply(command);
+        reply.ProtocolStatus = conversion.ProtocolStatus;
+        reply.Hresult = conversion.HResult;
+        reply.DiagnosticMessage = conversion.DiagnosticMessage;
+
+        return reply;
+    }
+
+    private static MxCommandReply CreateRejectedReply(
+        StaCommand command,
+        ProtocolStatusCode statusCode,
+        string message)
+    {
+        MxCommandReply reply = CreateBaseReply(command);
+        reply.ProtocolStatus = new ProtocolStatus
+        {
+            Code = statusCode,
+            Message = message,
+        };
+        reply.DiagnosticMessage = message;
+
+        return reply;
+    }
+
+    private static MxCommandReply CreateBaseReply(StaCommand command)
+    {
+        return new MxCommandReply
+        {
+            SessionId = command.SessionId,
+            CorrelationId = command.CorrelationId,
+            Kind = command.Kind,
+        };
+    }
+
+    private sealed class QueuedStaCommand
+    {
+        private readonly TaskCompletionSource<MxCommandReply> completion = new(
+            TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public QueuedStaCommand(StaCommand command)
+        {
+            Command = command;
+        }
+
+        public StaCommand Command { get; }
+
+        public Task<MxCommandReply> Task => completion.Task;
+
+        public void Complete(MxCommandReply reply)
+        {
+            completion.TrySetResult(reply);
+        }
+    }
+}

src/MxGateway.Worker/Sta/StaMessagePump.cs

@@ -0,0 +1,111 @@
+using System;
+using System.Runtime.InteropServices;
+using System.Threading;
+using Microsoft.Win32.SafeHandles;
+
+namespace MxGateway.Worker.Sta;
+
+public sealed class StaMessagePump
+{
+    private const uint Infinite = 0xFFFFFFFF;
+    private const uint MsgWaitFailed = 0xFFFFFFFF;
+    private const uint MwmoInputAvailable = 0x0004;
+    private const uint PmRemove = 0x0001;
+    private const uint QsAllInput = 0x04FF;
+
+    public void WaitForWorkOrMessages(WaitHandle commandWakeEvent, TimeSpan timeout)
+    {
+        if (commandWakeEvent is null)
+        {
+            throw new ArgumentNullException(nameof(commandWakeEvent));
+        }
+
+        uint timeoutMilliseconds = ToTimeoutMilliseconds(timeout);
+
+        SafeWaitHandle safeHandle = commandWakeEvent.SafeWaitHandle;
+        IntPtr[] handles = [safeHandle.DangerousGetHandle()];
+        uint result = MsgWaitForMultipleObjectsEx(
+            (uint)handles.Length,
+            handles,
+            timeoutMilliseconds,
+            QsAllInput,
+            MwmoInputAvailable);
+
+        if (result == MsgWaitFailed)
+        {
+            throw new InvalidOperationException(
+                "The worker STA message pump failed while waiting for command work or Windows messages.");
+        }
+    }
+
+    public int PumpPendingMessages()
+    {
+        int pumpedMessages = 0;
+
+        while (PeekMessage(out NativeMessage message, IntPtr.Zero, 0, 0, PmRemove))
+        {
+            TranslateMessage(ref message);
+            DispatchMessage(ref message);
+            pumpedMessages++;
+        }
+
+        return pumpedMessages;
+    }
+
+    private static uint ToTimeoutMilliseconds(TimeSpan timeout)
+    {
+        if (timeout == Timeout.InfiniteTimeSpan)
+        {
+            return Infinite;
+        }
+
+        if (timeout <= TimeSpan.Zero)
+        {
+            return 0;
+        }
+
+        return timeout.TotalMilliseconds >= uint.MaxValue
+            ? uint.MaxValue - 1
+            : (uint)Math.Ceiling(timeout.TotalMilliseconds);
+    }
+
+    [DllImport("user32.dll", SetLastError = true)]
+    private static extern uint MsgWaitForMultipleObjectsEx(
+        uint count,
+        IntPtr[] handles,
+        uint milliseconds,
+        uint wakeMask,
+        uint flags);
+
+    [DllImport("user32.dll", SetLastError = true)]
+    private static extern bool PeekMessage(
+        out NativeMessage message,
+        IntPtr windowHandle,
+        uint messageFilterMin,
+        uint messageFilterMax,
+        uint removeMessage);
+
+    [DllImport("user32.dll")]
+    private static extern bool TranslateMessage(ref NativeMessage message);
+
+    [DllImport("user32.dll")]
+    private static extern IntPtr DispatchMessage(ref NativeMessage message);
+
+    [StructLayout(LayoutKind.Sequential)]
+    private struct NativeMessage
+    {
+        public IntPtr WindowHandle;
+        public uint Message;
+        public UIntPtr WParam;
+        public IntPtr LParam;
+        public uint Time;
+        public NativePoint Point;
+    }
+
+    [StructLayout(LayoutKind.Sequential)]
+    private struct NativePoint
+    {
+        public int X;
+        public int Y;
+    }
+}

src/MxGateway.Worker/Sta/StaRuntime.cs

@@ -0,0 +1,267 @@
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MxGateway.Worker.Sta;
+
+public sealed class StaRuntime : IDisposable
+{
+    private readonly IStaComApartmentInitializer comApartmentInitializer;
+    private readonly StaMessagePump messagePump;
+    private readonly ConcurrentQueue<IStaWorkItem> commandQueue = new();
+    private readonly AutoResetEvent commandWakeEvent = new(false);
+    private readonly ManualResetEventSlim startedEvent = new(false);
+    private readonly ManualResetEventSlim stoppedEvent = new(false);
+    private readonly object gate = new();
+    private readonly Thread staThread;
+    private readonly TimeSpan idlePumpInterval;
+    private bool disposed;
+    private bool startRequested;
+    private bool shutdownRequested;
+    private Exception? startupException;
+    private long lastActivityUtcTicks;
+    private bool comInitialized;
+
+    public StaRuntime()
+        : this(new StaComApartmentInitializer(), new StaMessagePump(), TimeSpan.FromMilliseconds(50))
+    {
+    }
+
+    public StaRuntime(
+        IStaComApartmentInitializer comApartmentInitializer,
+        StaMessagePump messagePump,
+        TimeSpan idlePumpInterval)
+    {
+        this.comApartmentInitializer = comApartmentInitializer
+            ?? throw new ArgumentNullException(nameof(comApartmentInitializer));
+        this.messagePump = messagePump ?? throw new ArgumentNullException(nameof(messagePump));
+
+        if (idlePumpInterval <= TimeSpan.Zero)
+        {
+            throw new ArgumentOutOfRangeException(
+                nameof(idlePumpInterval),
+                "The idle pump interval must be greater than zero.");
+        }
+
+        this.idlePumpInterval = idlePumpInterval;
+        lastActivityUtcTicks = DateTimeOffset.UtcNow.UtcTicks;
+        staThread = new Thread(ThreadMain)
+        {
+            IsBackground = true,
+            Name = "MxGateway.Worker.STA"
+        };
+        staThread.SetApartmentState(ApartmentState.STA);
+    }
+
+    public int? StaThreadId { get; private set; }
+
+    public DateTimeOffset LastActivityUtc =>
+        new(new DateTime(Volatile.Read(ref lastActivityUtcTicks), DateTimeKind.Utc));
+
+    public bool IsRunning => startedEvent.IsSet && !stoppedEvent.IsSet;
+
+    public void Start()
+    {
+        ThrowIfDisposed();
+
+        lock (gate)
+        {
+            if (shutdownRequested)
+            {
+                throw new InvalidOperationException("The worker STA runtime is shutting down.");
+            }
+
+            if (!startRequested)
+            {
+                startRequested = true;
+                staThread.Start();
+            }
+        }
+
+        startedEvent.Wait();
+        if (startupException is not null)
+        {
+            throw new InvalidOperationException(
+                "The worker STA runtime failed to initialize.",
+                startupException);
+        }
+    }
+
+    public Task InvokeAsync(Action command, CancellationToken cancellationToken = default)
+    {
+        if (command is null)
+        {
+            throw new ArgumentNullException(nameof(command));
+        }
+
+        return InvokeAsync(
+            () =>
+            {
+                command();
+                return true;
+            },
+            cancellationToken);
+    }
+
+    public Task<T> InvokeAsync<T>(Func<T> command, CancellationToken cancellationToken = default)
+    {
+        if (command is null)
+        {
+            throw new ArgumentNullException(nameof(command));
+        }
+
+        ThrowIfDisposed();
+
+        if (cancellationToken.IsCancellationRequested)
+        {
+            return Task.FromCanceled<T>(cancellationToken);
+        }
+
+        StaWorkItem<T> workItem = new(command, cancellationToken);
+
+        lock (gate)
+        {
+            if (shutdownRequested)
+            {
+                return Task.FromException<T>(
+                    new InvalidOperationException("The worker STA runtime is shutting down."));
+            }
+
+            commandQueue.Enqueue(workItem);
+        }
+
+        commandWakeEvent.Set();
+        return workItem.Task;
+    }
+
+    public bool Shutdown(TimeSpan timeout)
+    {
+        if (timeout < TimeSpan.Zero && timeout != Timeout.InfiniteTimeSpan)
+        {
+            throw new ArgumentOutOfRangeException(nameof(timeout));
+        }
+
+        lock (gate)
+        {
+            shutdownRequested = true;
+        }
+
+        commandWakeEvent.Set();
+
+        if (!startedEvent.IsSet && !staThread.IsAlive)
+        {
+            CancelQueuedCommands();
+            stoppedEvent.Set();
+            return true;
+        }
+
+        bool stopped = stoppedEvent.Wait(timeout);
+        if (stopped)
+        {
+            CancelQueuedCommands();
+        }
+
+        return stopped;
+    }
+
+    public void Dispose()
+    {
+        if (disposed)
+        {
+            return;
+        }
+
+        bool stopped = Shutdown(TimeSpan.FromSeconds(5));
+        if (stopped)
+        {
+            commandWakeEvent.Dispose();
+            startedEvent.Dispose();
+            stoppedEvent.Dispose();
+        }
+
+        disposed = true;
+    }
+
+    private void ThreadMain()
+    {
+        try
+        {
+            StaThreadId = Thread.CurrentThread.ManagedThreadId;
+            comApartmentInitializer.Initialize();
+            comInitialized = true;
+            MarkActivity();
+            startedEvent.Set();
+
+            while (!IsShutdownRequested())
+            {
+                ProcessQueuedCommands();
+                messagePump.WaitForWorkOrMessages(commandWakeEvent, idlePumpInterval);
+                messagePump.PumpPendingMessages();
+                MarkActivity();
+            }
+
+            ProcessQueuedCommands();
+        }
+        catch (Exception exception)
+        {
+            startupException = exception;
+            startedEvent.Set();
+        }
+        finally
+        {
+            CancelQueuedCommands();
+            try
+            {
+                if (comInitialized)
+                {
+                    comApartmentInitializer.Uninitialize();
+                }
+            }
+            finally
+            {
+                MarkActivity();
+                stoppedEvent.Set();
+            }
+        }
+    }
+
+    private void ProcessQueuedCommands()
+    {
+        while (commandQueue.TryDequeue(out IStaWorkItem? workItem))
+        {
+            MarkActivity();
+            workItem.Execute();
+            MarkActivity();
+        }
+    }
+
+    private void CancelQueuedCommands()
+    {
+        while (commandQueue.TryDequeue(out IStaWorkItem? workItem))
+        {
+            workItem.CancelBeforeExecution();
+        }
+    }
+
+    private bool IsShutdownRequested()
+    {
+        lock (gate)
+        {
+            return shutdownRequested;
+        }
+    }
+
+    private void MarkActivity()
+    {
+        Volatile.Write(ref lastActivityUtcTicks, DateTimeOffset.UtcNow.UtcTicks);
+    }
+
+    private void ThrowIfDisposed()
+    {
+        if (disposed)
+        {
+            throw new ObjectDisposedException(nameof(StaRuntime));
+        }
+    }
+}

src/MxGateway.Worker/Sta/StaWorkItem.cs

@@ -0,0 +1,71 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MxGateway.Worker.Sta;
+
+internal sealed class StaWorkItem<T> : IStaWorkItem
+{
+    private readonly Func<T> command;
+    private readonly CancellationToken cancellationToken;
+    private readonly CancellationTokenRegistration cancellationRegistration;
+    private int started;
+
+    public StaWorkItem(Func<T> command, CancellationToken cancellationToken)
+    {
+        this.command = command ?? throw new ArgumentNullException(nameof(command));
+        this.cancellationToken = cancellationToken;
+        Completion = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        if (cancellationToken.CanBeCanceled)
+        {
+            cancellationRegistration = cancellationToken.Register(
+                () =>
+                {
+                    if (Interlocked.CompareExchange(ref started, 1, 0) == 0)
+                    {
+                        Completion.TrySetCanceled(cancellationToken);
+                    }
+                });
+        }
+    }
+
+    public Task<T> Task => Completion.Task;
+
+    private TaskCompletionSource<T> Completion { get; }
+
+    public void CancelBeforeExecution()
+    {
+        if (Interlocked.CompareExchange(ref started, 1, 0) == 0)
+        {
+            Completion.TrySetCanceled(cancellationToken);
+            cancellationRegistration.Dispose();
+        }
+    }
+
+    public void Execute()
+    {
+        if (Interlocked.CompareExchange(ref started, 1, 0) != 0)
+        {
+            cancellationRegistration.Dispose();
+            return;
+        }
+
+        cancellationRegistration.Dispose();
+
+        if (cancellationToken.IsCancellationRequested)
+        {
+            Completion.TrySetCanceled(cancellationToken);
+            return;
+        }
+
+        try
+        {
+            Completion.TrySetResult(command());
+        }
+        catch (Exception exception)
+        {
+            Completion.TrySetException(exception);
+        }
+    }
+}