Issue #12: implement session manager and registry

Verified with dotnet build src\\MxGateway.sln, dotnet test src\\MxGateway.Worker.Tests\\MxGateway.Worker.Tests.csproj -p:Platform=x86, and dotnet test src\\MxGateway.sln --no-build.

docs/gateway-process-design.md

@@ -330,6 +330,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:

gateway.md

@@ -799,6 +799,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

src/MxGateway.Server/GatewayApplication.cs

@@ -4,6 +4,7 @@ using MxGateway.Server.Diagnostics;
 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;
@@ -31,6 +32,7 @@ public static class GatewayApplication
         builder.Services.AddHealthChecks();
         builder.Services.AddSingleton<GatewayMetrics>();
         builder.Services.AddWorkerProcessLauncher();
+        builder.Services.AddGatewaySessions();
 
         return builder;
     }

src/MxGateway.Server/Sessions/GatewaySession.cs

@@ -0,0 +1,290 @@
+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;
+
+    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 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 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;
+        }
+    }
+}

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,10 @@
+namespace MxGateway.Server.Sessions;
+
+public enum SessionManagerErrorCode
+{
+    SessionNotFound,
+    SessionNotReady,
+    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.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.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/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}'.";
+    }
+}