Implement dashboard authentication

docs/gateway-dashboard-design.md

@@ -257,19 +257,18 @@ Do not show API key secrets or pepper values.
 
 ## Authentication And Authorization
 
-Dashboard access should use the same API-key authentication model as gRPC where
+Dashboard access uses the same API-key authentication model as gRPC where
 practical.
 
-Recommended v1 behavior:
+Implemented v1 behavior:
 
-- dashboard disabled by default unless configured,
 - when enabled, require API key auth,
 - require `admin` scope for dashboard access,
-- accept API key through a secure cookie established by a simple login form, or
-  through reverse-proxy/header configuration for local deployments,
-- do not put API keys in query strings.
+- accept API key through a secure cookie established by a simple login form,
+- do not put API keys in query strings,
+- validate anti-forgery tokens for login and logout posts.
 
-Simplest implementation path:
+The implementation path is:
 
 1. Add `/dashboard/login`.
 2. User submits API key over HTTPS.
@@ -281,6 +280,13 @@ Simplest implementation path:
 For local development, allow an explicit `Dashboard:AllowAnonymousLocalhost`
 option. It must default to false.
 
+`DashboardAuthenticator` keeps API-key validation outside UI components. It
+formats the submitted key as a bearer authorization header for
+`IApiKeyVerifier`, rejects non-admin keys when `Dashboard:RequireAdminScope` is
+enabled, and creates the dashboard cookie principal without storing raw API key
+material. `DashboardAuthorizationHandler` enforces the cookie, admin-scope, and
+explicit loopback bypass decisions for all protected dashboard routes.
+
 ## Configuration
 
 Suggested configuration:

docs/gateway-process-design.md

@@ -206,23 +206,13 @@ accounting and a clear fan-out policy.
 Behavior:
 
 1. Validate session id and authorize event access.
-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
+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
    preserve the required details.
 
-`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.
+The gateway must not reorder events from one worker.
 
 ## Web Dashboard
 
@@ -594,8 +584,7 @@ worker MXAccess event
   -> worker outbound event queue
   -> worker pipe writer
   -> gateway read loop
-  -> worker client event queue
-  -> EventStreamService bounded stream queue
+  -> session event channel
   -> gRPC StreamEvents
 ```
 
@@ -609,15 +598,13 @@ The gateway should record:
 
 Default backpressure policy for parity testing should be fail-fast:
 
-1. If the worker client event queue fills, fault the worker client.
-2. If the public stream queue fills, fault the gateway session.
+1. If the session event channel fills, fault the 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. `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.
+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.
 
 The gateway should not synthesize `OperationComplete` from write completion,
 command replies, ASB completion queues, or completion-only status frames. Forward
@@ -663,6 +650,16 @@ server-streaming calls and stores the authenticated `ApiKeyIdentity` in
 `Authentication:Mode` set to `Disabled` bypasses API-key verification for local
 development only.
 
+Dashboard authentication reuses the API-key verifier and scope model. The
+dashboard login endpoint accepts the key in a form post, checks `admin` scope
+when `Dashboard:RequireAdminScope` is enabled, and signs in with the
+`MxGateway.Dashboard` cookie scheme. The cookie is HTTP-only, secure, strict
+SameSite, and scoped with the `__Host-MxGatewayDashboard` name. Logout clears
+that cookie. Login and logout posts use anti-forgery validation, and dashboard
+API keys are not accepted in query strings. `Dashboard:AllowAnonymousLocalhost`
+allows only loopback requests to bypass the dashboard cookie requirement and
+defaults to `false`.
+
 Recommended scopes:
 
 - `session:open`

gateway.md

@@ -109,6 +109,13 @@ 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.
+Dashboard routes use the same API-key verifier as gRPC. `/dashboard/login`
+accepts the API key in a form body, validates the configured `admin` scope,
+and issues an HTTP-only secure cookie for subsequent dashboard requests.
+`/dashboard/logout` clears that cookie. Login and logout posts validate
+anti-forgery tokens, and API keys are never accepted through query strings.
+`Dashboard:AllowAnonymousLocalhost` can bypass the cookie requirement for
+loopback requests only when explicitly enabled.
 
 ### Worker Process
 
@@ -520,7 +527,11 @@ Worker policy:
 
 - bounded outbound event channel,
 - never block MXAccess event handler on pipe writes,
-- fail the worker session when the outbound channel is full.
+- 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.
 
 For full parity testing, default should be fail-fast rather than silent drop.
 For production high-rate telemetry, add explicit coalescing modes.
@@ -529,15 +540,9 @@ Gateway policy:
 
 - one event sequencer per session,
 - preserve per-session event order,
-- 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.
+- support multiple client event subscribers only if explicitly required,
+- apply backpressure from slow gRPC streams,
+- disconnect or coalesce according to client-selected mode.
 
 ## Isolation And Fault Handling
 
@@ -859,11 +864,10 @@ 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.
+using explicit mapper code for public-to-worker commands, worker replies, and
+events. 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
 

src/MxGateway.Server/Dashboard/DashboardAuthenticationDefaults.cs

@@ -0,0 +1,10 @@
+namespace MxGateway.Server.Dashboard;
+
+public static class DashboardAuthenticationDefaults
+{
+    public const string AuthenticationScheme = "MxGateway.Dashboard";
+    public const string AuthorizationPolicy = "MxGateway.Dashboard";
+    public const string ScopeClaimType = "scope";
+    public const string KeyPrefixClaimType = "mxgateway:key_prefix";
+    public const string CookieName = "__Host-MxGatewayDashboard";
+}

src/MxGateway.Server/Dashboard/DashboardAuthenticationResult.cs

@@ -0,0 +1,19 @@
+using System.Security.Claims;
+
+namespace MxGateway.Server.Dashboard;
+
+public sealed record DashboardAuthenticationResult(
+    bool Succeeded,
+    ClaimsPrincipal? Principal,
+    string? FailureMessage)
+{
+    public static DashboardAuthenticationResult Success(ClaimsPrincipal principal)
+    {
+        return new DashboardAuthenticationResult(true, principal, null);
+    }
+
+    public static DashboardAuthenticationResult Fail(string failureMessage)
+    {
+        return new DashboardAuthenticationResult(false, null, failureMessage);
+    }
+}

src/MxGateway.Server/Dashboard/DashboardAuthenticator.cs

@@ -0,0 +1,81 @@
+using System.Security.Claims;
+using Microsoft.Extensions.Options;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Security.Authentication;
+using MxGateway.Server.Security.Authorization;
+
+namespace MxGateway.Server.Dashboard;
+
+public sealed class DashboardAuthenticator(
+    IApiKeyVerifier apiKeyVerifier,
+    IOptions<GatewayOptions> options) : IDashboardAuthenticator
+{
+    private const string GenericFailureMessage = "The API key is invalid or is not authorized for dashboard access.";
+
+    public async Task<DashboardAuthenticationResult> AuthenticateAsync(
+        string? apiKey,
+        CancellationToken cancellationToken)
+    {
+        if (options.Value.Authentication.Mode == AuthenticationMode.Disabled)
+        {
+            return DashboardAuthenticationResult.Success(CreatePrincipal(new ApiKeyIdentity(
+                KeyId: "authentication-disabled",
+                KeyPrefix: "authentication-disabled",
+                DisplayName: "Authentication Disabled",
+                Scopes: new HashSet<string>([GatewayScopes.Admin], StringComparer.Ordinal))));
+        }
+
+        if (string.IsNullOrWhiteSpace(apiKey))
+        {
+            return DashboardAuthenticationResult.Fail(GenericFailureMessage);
+        }
+
+        ApiKeyVerificationResult verificationResult = await apiKeyVerifier
+            .VerifyAsync(FormatAuthorizationHeader(apiKey), cancellationToken)
+            .ConfigureAwait(false);
+
+        if (!verificationResult.Succeeded || verificationResult.Identity is null)
+        {
+            return DashboardAuthenticationResult.Fail(GenericFailureMessage);
+        }
+
+        if (options.Value.Dashboard.RequireAdminScope
+            && !verificationResult.Identity.Scopes.Contains(GatewayScopes.Admin))
+        {
+            return DashboardAuthenticationResult.Fail(GenericFailureMessage);
+        }
+
+        return DashboardAuthenticationResult.Success(CreatePrincipal(verificationResult.Identity));
+    }
+
+    private static string FormatAuthorizationHeader(string apiKey)
+    {
+        string trimmedApiKey = apiKey.Trim();
+
+        return trimmedApiKey.StartsWith("Bearer ", StringComparison.OrdinalIgnoreCase)
+            ? trimmedApiKey
+            : $"Bearer {trimmedApiKey}";
+    }
+
+    private static ClaimsPrincipal CreatePrincipal(ApiKeyIdentity identity)
+    {
+        List<Claim> claims =
+        [
+            new Claim(ClaimTypes.NameIdentifier, identity.KeyId),
+            new Claim(ClaimTypes.Name, identity.DisplayName),
+            new Claim(DashboardAuthenticationDefaults.KeyPrefixClaimType, identity.KeyPrefix)
+        ];
+
+        claims.AddRange(identity.Scopes.Select(scope => new Claim(
+            DashboardAuthenticationDefaults.ScopeClaimType,
+            scope)));
+
+        ClaimsIdentity claimsIdentity = new(
+            claims,
+            DashboardAuthenticationDefaults.AuthenticationScheme,
+            ClaimTypes.Name,
+            DashboardAuthenticationDefaults.ScopeClaimType);
+
+        return new ClaimsPrincipal(claimsIdentity);
+    }
+}

src/MxGateway.Server/Dashboard/DashboardAuthorizationHandler.cs

@@ -0,0 +1,59 @@
+using System.Net;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.Extensions.Options;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Security.Authorization;
+
+namespace MxGateway.Server.Dashboard;
+
+public sealed class DashboardAuthorizationHandler(
+    IHttpContextAccessor httpContextAccessor,
+    IOptions<GatewayOptions> options) : AuthorizationHandler<DashboardAuthorizationRequirement>
+{
+    protected override Task HandleRequirementAsync(
+        AuthorizationHandlerContext context,
+        DashboardAuthorizationRequirement requirement)
+    {
+        GatewayOptions gatewayOptions = options.Value;
+
+        if (gatewayOptions.Authentication.Mode == AuthenticationMode.Disabled)
+        {
+            context.Succeed(requirement);
+
+            return Task.CompletedTask;
+        }
+
+        if (gatewayOptions.Dashboard.AllowAnonymousLocalhost && IsLoopbackRequest())
+        {
+            context.Succeed(requirement);
+
+            return Task.CompletedTask;
+        }
+
+        if (context.User.Identity?.IsAuthenticated != true)
+        {
+            return Task.CompletedTask;
+        }
+
+        if (!gatewayOptions.Dashboard.RequireAdminScope || HasAdminScope(context))
+        {
+            context.Succeed(requirement);
+        }
+
+        return Task.CompletedTask;
+    }
+
+    private bool IsLoopbackRequest()
+    {
+        IPAddress? remoteAddress = httpContextAccessor.HttpContext?.Connection.RemoteIpAddress;
+
+        return remoteAddress is not null && IPAddress.IsLoopback(remoteAddress);
+    }
+
+    private static bool HasAdminScope(AuthorizationHandlerContext context)
+    {
+        return context.User.HasClaim(
+            DashboardAuthenticationDefaults.ScopeClaimType,
+            GatewayScopes.Admin);
+    }
+}

src/MxGateway.Server/Dashboard/DashboardAuthorizationRequirement.cs

@@ -0,0 +1,5 @@
+using Microsoft.AspNetCore.Authorization;
+
+namespace MxGateway.Server.Dashboard;
+
+public sealed class DashboardAuthorizationRequirement : IAuthorizationRequirement;

src/MxGateway.Server/Dashboard/DashboardEndpointRouteBuilderExtensions.cs

@@ -0,0 +1,217 @@
+using System.Text.Encodings.Web;
+using Microsoft.AspNetCore.Antiforgery;
+using Microsoft.AspNetCore.Authentication;
+using Microsoft.AspNetCore.Http.HttpResults;
+using MxGateway.Server.Configuration;
+
+namespace MxGateway.Server.Dashboard;
+
+public static class DashboardEndpointRouteBuilderExtensions
+{
+    public static IEndpointRouteBuilder MapGatewayDashboard(this IEndpointRouteBuilder endpoints)
+    {
+        IConfiguration configuration = endpoints.ServiceProvider.GetRequiredService<IConfiguration>();
+        IConfigurationSection dashboardSection = configuration
+            .GetSection($"{GatewayOptions.SectionName}:Dashboard");
+
+        if (bool.TryParse(dashboardSection["Enabled"], out bool enabled) && !enabled)
+        {
+            return endpoints;
+        }
+
+        string pathBase = NormalizePathBase(dashboardSection["PathBase"] ?? new DashboardOptions().PathBase);
+        RouteGroupBuilder dashboard = endpoints.MapGroup(pathBase);
+
+        dashboard.MapGet(
+                "/",
+                (HttpContext httpContext, IAntiforgery antiforgery, IDashboardSnapshotService snapshotService) =>
+                    GetDashboardHomeAsync(httpContext, antiforgery, snapshotService, pathBase))
+            .RequireAuthorization(DashboardAuthenticationDefaults.AuthorizationPolicy)
+            .WithName("DashboardHome");
+
+        dashboard.MapGet(
+                "/login",
+                (HttpContext httpContext, IAntiforgery antiforgery) => GetLoginAsync(httpContext, antiforgery, pathBase))
+            .AllowAnonymous()
+            .WithName("DashboardLogin");
+
+        dashboard.MapPost(
+                "/login",
+                (HttpContext httpContext, IAntiforgery antiforgery, IDashboardAuthenticator authenticator) =>
+                    PostLoginAsync(httpContext, antiforgery, authenticator, pathBase))
+            .AllowAnonymous()
+            .WithName("DashboardLoginPost");
+
+        dashboard.MapPost(
+                "/logout",
+                (HttpContext httpContext, IAntiforgery antiforgery) => PostLogoutAsync(httpContext, antiforgery, pathBase))
+            .RequireAuthorization(DashboardAuthenticationDefaults.AuthorizationPolicy)
+            .WithName("DashboardLogout");
+
+        dashboard.MapGet("/denied", () => Results.Content(
+                RenderPage("Access denied", "<p>The signed-in API key is not authorized for dashboard access.</p>"),
+                "text/html"))
+            .AllowAnonymous()
+            .WithName("DashboardAccessDenied");
+
+        return endpoints;
+    }
+
+    private static ContentHttpResult GetDashboardHomeAsync(
+        HttpContext httpContext,
+        IAntiforgery antiforgery,
+        IDashboardSnapshotService snapshotService,
+        string pathBase)
+    {
+        AntiforgeryTokenSet tokens = antiforgery.GetAndStoreTokens(httpContext);
+        DashboardSnapshot snapshot = snapshotService.GetSnapshot();
+        string requestToken = tokens.RequestToken ?? string.Empty;
+        string body = $"""
+            <form method="post" action="{HtmlEncoder.Default.Encode(pathBase + "/logout")}" class="mb-3">
+                <input name="{tokens.FormFieldName}" type="hidden" value="{HtmlEncoder.Default.Encode(requestToken)}" />
+                <button type="submit">Sign out</button>
+            </form>
+            <dl>
+                <dt>Open sessions</dt>
+                <dd>{snapshot.Sessions.Count}</dd>
+                <dt>Workers</dt>
+                <dd>{snapshot.Workers.Count}</dd>
+                <dt>Faults</dt>
+                <dd>{snapshot.Faults.Count}</dd>
+            </dl>
+            """;
+
+        return TypedResults.Content(RenderPage("MXAccess Gateway Dashboard", body), "text/html");
+    }
+
+    private static Task<ContentHttpResult> GetLoginAsync(
+        HttpContext httpContext,
+        IAntiforgery antiforgery,
+        string pathBase)
+    {
+        string returnUrl = SanitizeReturnUrl(
+            httpContext.Request.Query["returnUrl"].ToString(),
+            pathBase);
+
+        return Task.FromResult(TypedResults.Content(
+            RenderLoginPage(httpContext, antiforgery, returnUrl, pathBase, failureMessage: null),
+            "text/html"));
+    }
+
+    private static async Task<IResult> PostLoginAsync(
+        HttpContext httpContext,
+        IAntiforgery antiforgery,
+        IDashboardAuthenticator authenticator,
+        string pathBase)
+    {
+        await antiforgery.ValidateRequestAsync(httpContext).ConfigureAwait(false);
+
+        IFormCollection form = await httpContext.Request
+            .ReadFormAsync(httpContext.RequestAborted)
+            .ConfigureAwait(false);
+        string returnUrl = SanitizeReturnUrl(
+            form["returnUrl"].ToString(),
+            pathBase);
+
+        DashboardAuthenticationResult result = await authenticator
+            .AuthenticateAsync(form["apiKey"].ToString(), httpContext.RequestAborted)
+            .ConfigureAwait(false);
+
+        if (!result.Succeeded || result.Principal is null)
+        {
+            return TypedResults.Content(
+                RenderLoginPage(httpContext, antiforgery, returnUrl, pathBase, result.FailureMessage),
+                "text/html",
+                statusCode: StatusCodes.Status401Unauthorized);
+        }
+
+        await httpContext
+            .SignInAsync(DashboardAuthenticationDefaults.AuthenticationScheme, result.Principal)
+            .ConfigureAwait(false);
+
+        return Results.LocalRedirect(returnUrl);
+    }
+
+    private static async Task<IResult> PostLogoutAsync(
+        HttpContext httpContext,
+        IAntiforgery antiforgery,
+        string pathBase)
+    {
+        await antiforgery.ValidateRequestAsync(httpContext).ConfigureAwait(false);
+        await httpContext
+            .SignOutAsync(DashboardAuthenticationDefaults.AuthenticationScheme)
+            .ConfigureAwait(false);
+
+        return Results.LocalRedirect($"{pathBase}/login");
+    }
+
+    private static string RenderLoginPage(
+        HttpContext httpContext,
+        IAntiforgery antiforgery,
+        string returnUrl,
+        string pathBase,
+        string? failureMessage)
+    {
+        AntiforgeryTokenSet tokens = antiforgery.GetAndStoreTokens(httpContext);
+        string requestToken = tokens.RequestToken ?? string.Empty;
+        string alert = string.IsNullOrWhiteSpace(failureMessage)
+            ? string.Empty
+            : $"<p role=\"alert\">{HtmlEncoder.Default.Encode(failureMessage)}</p>";
+
+        string body = $"""
+            {alert}
+            <form method="post" action="{HtmlEncoder.Default.Encode(pathBase + "/login")}">
+                <input name="{tokens.FormFieldName}" type="hidden" value="{HtmlEncoder.Default.Encode(requestToken)}" />
+                <input name="returnUrl" type="hidden" value="{HtmlEncoder.Default.Encode(returnUrl)}" />
+                <label for="apiKey">API key</label>
+                <input id="apiKey" name="apiKey" type="password" autocomplete="off" />
+                <button type="submit">Sign in</button>
+            </form>
+            """;
+
+        return RenderPage("Dashboard Sign In", body);
+    }
+
+    private static string RenderPage(string title, string body)
+    {
+        return $"""
+            <!doctype html>
+            <html lang="en">
+            <head>
+                <meta charset="utf-8" />
+                <meta name="viewport" content="width=device-width, initial-scale=1" />
+                <title>{HtmlEncoder.Default.Encode(title)}</title>
+            </head>
+            <body>
+                <main>
+                    <h1>{HtmlEncoder.Default.Encode(title)}</h1>
+                    {body}
+                </main>
+            </body>
+            </html>
+            """;
+    }
+
+    private static string NormalizePathBase(string pathBase)
+    {
+        string normalized = pathBase.TrimEnd('/');
+
+        return string.IsNullOrWhiteSpace(normalized) || !normalized.StartsWith("/", StringComparison.Ordinal)
+            ? "/dashboard"
+            : normalized;
+    }
+
+    private static string SanitizeReturnUrl(string? returnUrl, string pathBase)
+    {
+        if (string.IsNullOrWhiteSpace(returnUrl)
+            || !returnUrl.StartsWith("/", StringComparison.Ordinal)
+            || returnUrl.StartsWith("//", StringComparison.Ordinal)
+            || !returnUrl.StartsWith(pathBase, StringComparison.OrdinalIgnoreCase)
+            || Uri.TryCreate(returnUrl, UriKind.Absolute, out _))
+        {
+            return pathBase;
+        }
+
+        return returnUrl;
+    }
+}

src/MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs

@@ -1,3 +1,8 @@
+using Microsoft.AspNetCore.Authentication.Cookies;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.Extensions.Options;
+using MxGateway.Server.Configuration;
+
 namespace MxGateway.Server.Dashboard;
 
 public static class DashboardServiceCollectionExtensions
@@ -5,7 +10,44 @@ public static class DashboardServiceCollectionExtensions
     public static IServiceCollection AddGatewayDashboard(this IServiceCollection services)
     {
         services.AddSingleton<IDashboardSnapshotService, DashboardSnapshotService>();
+        services.AddSingleton<IDashboardAuthenticator, DashboardAuthenticator>();
+        services.AddHttpContextAccessor();
+        services.AddAntiforgery();
+        services
+            .AddAuthentication(DashboardAuthenticationDefaults.AuthenticationScheme)
+            .AddCookie(DashboardAuthenticationDefaults.AuthenticationScheme);
+        services.AddOptions<CookieAuthenticationOptions>(DashboardAuthenticationDefaults.AuthenticationScheme)
+            .Configure<IOptions<GatewayOptions>>(ConfigureCookieOptions);
+        services.AddAuthorization(options =>
+        {
+            options.AddPolicy(
+                DashboardAuthenticationDefaults.AuthorizationPolicy,
+                policy => policy.AddRequirements(new DashboardAuthorizationRequirement()));
+        });
+        services.AddSingleton<IAuthorizationHandler, DashboardAuthorizationHandler>();
 
         return services;
     }
+
+    private static void ConfigureCookieOptions(
+        CookieAuthenticationOptions cookieOptions,
+        IOptions<GatewayOptions> gatewayOptions)
+    {
+        string pathBase = gatewayOptions.Value.Dashboard.PathBase.TrimEnd('/');
+        if (string.IsNullOrWhiteSpace(pathBase))
+        {
+            pathBase = "/dashboard";
+        }
+
+        cookieOptions.Cookie.Name = DashboardAuthenticationDefaults.CookieName;
+        cookieOptions.Cookie.HttpOnly = true;
+        cookieOptions.Cookie.SecurePolicy = CookieSecurePolicy.Always;
+        cookieOptions.Cookie.SameSite = SameSiteMode.Strict;
+        cookieOptions.Cookie.Path = "/";
+        cookieOptions.LoginPath = $"{pathBase}/login";
+        cookieOptions.LogoutPath = $"{pathBase}/logout";
+        cookieOptions.AccessDeniedPath = $"{pathBase}/denied";
+        cookieOptions.ExpireTimeSpan = TimeSpan.FromHours(8);
+        cookieOptions.SlidingExpiration = true;
+    }
 }

src/MxGateway.Server/Dashboard/IDashboardAuthenticator.cs

@@ -0,0 +1,8 @@
+namespace MxGateway.Server.Dashboard;
+
+public interface IDashboardAuthenticator
+{
+    Task<DashboardAuthenticationResult> AuthenticateAsync(
+        string? apiKey,
+        CancellationToken cancellationToken);
+}

src/MxGateway.Server/GatewayApplication.cs

@@ -19,6 +19,8 @@ public static class GatewayApplication
         WebApplication app = builder.Build();
 
         app.UseGatewayRequestLoggingScope();
+        app.UseAuthentication();
+        app.UseAuthorization();
         app.MapGatewayEndpoints();
 
         return app;
@@ -35,7 +37,6 @@ public static class GatewayApplication
         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();
@@ -56,6 +57,7 @@ public static class GatewayApplication
             .WithName("LiveHealth");
 
         endpoints.MapGrpcService<MxAccessGatewayService>();
+        endpoints.MapGatewayDashboard();
 
         return endpoints;
     }

src/MxGateway.Server/Grpc/EventStreamService.cs

@@ -1,140 +0,0 @@
-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

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

src/MxGateway.Server/Grpc/MxAccessGatewayService.cs

@@ -12,7 +12,6 @@ public sealed class MxAccessGatewayService(
     IGatewayRequestIdentityAccessor identityAccessor,
     MxAccessGrpcRequestValidator requestValidator,
     MxAccessGrpcMapper mapper,
-    IEventStreamService eventStreamService,
     ILogger<MxAccessGatewayService> logger) : MxAccessGateway.MxAccessGatewayBase
 {
     public override async Task<OpenSessionReply> OpenSession(
@@ -103,11 +102,17 @@ public sealed class MxAccessGatewayService(
         try
         {
             requestValidator.ValidateStreamEvents(request);
-            await foreach (MxEvent publicEvent in eventStreamService
-                .StreamEventsAsync(request, context.CancellationToken)
+            await foreach (WorkerEvent workerEvent in sessionManager
+                .ReadEventsAsync(request.SessionId, context.CancellationToken)
                 .WithCancellation(context.CancellationToken)
                 .ConfigureAwait(false))
             {
+                MxEvent publicEvent = mapper.MapEvent(workerEvent);
+                if (publicEvent.WorkerSequence <= request.AfterWorkerSequence)
+                {
+                    continue;
+                }
+
                 await responseStream.WriteAsync(publicEvent).ConfigureAwait(false);
             }
         }
@@ -149,8 +154,6 @@ public sealed class MxAccessGatewayService(
         {
             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,

src/MxGateway.Server/Sessions/GatewaySession.cs

@@ -13,7 +13,6 @@ public sealed class GatewaySession
     private DateTimeOffset _lastClientActivityAt;
     private DateTimeOffset? _leaseExpiresAt;
     private bool _closeStarted;
-    private int _activeEventSubscriberCount;
 
     public GatewaySession(
         string sessionId,
@@ -132,17 +131,6 @@ public sealed class GatewaySession
         }
     }
 
-    public int ActiveEventSubscriberCount
-    {
-        get
-        {
-            lock (_syncRoot)
-            {
-                return _activeEventSubscriberCount;
-            }
-        }
-    }
-
     public void AttachWorkerClient(IWorkerClient workerClient)
     {
         ArgumentNullException.ThrowIfNull(workerClient);
@@ -214,29 +202,6 @@ public sealed class GatewaySession
         }
     }
 
-    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)
@@ -322,31 +287,4 @@ public sealed class GatewaySession
             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/SessionManagerErrorCode.cs

@@ -4,8 +4,6 @@ public enum SessionManagerErrorCode
 {
     SessionNotFound,
     SessionNotReady,
-    EventSubscriberAlreadyActive,
-    EventQueueOverflow,
     SessionLimitExceeded,
     OpenFailed,
     CloseFailed,

src/MxGateway.Server/Workers/WorkerClient.cs

@@ -29,7 +29,6 @@ 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;
@@ -198,8 +197,6 @@ 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;
         }
     }
@@ -397,6 +394,11 @@ 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");
@@ -404,11 +406,7 @@ 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/DashboardAuthenticatorTests.cs

@@ -0,0 +1,113 @@
+using System.Security.Claims;
+using Microsoft.Extensions.Options;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Dashboard;
+using MxGateway.Server.Security.Authentication;
+using MxGateway.Server.Security.Authorization;
+
+namespace MxGateway.Tests.Gateway.Dashboard;
+
+public sealed class DashboardAuthenticatorTests
+{
+    [Fact]
+    public async Task AuthenticateAsync_AdminKey_ReturnsCookiePrincipal()
+    {
+        FakeApiKeyVerifier verifier = new(SuccessWithScopes(GatewayScopes.Admin));
+        DashboardAuthenticator authenticator = CreateAuthenticator(verifier);
+
+        DashboardAuthenticationResult result = await authenticator.AuthenticateAsync(
+            "mxgw_operator01_super-secret",
+            CancellationToken.None);
+
+        Assert.True(result.Succeeded);
+        Assert.NotNull(result.Principal);
+        Assert.Equal("operator01", result.Principal.FindFirst(ClaimTypes.NameIdentifier)?.Value);
+        Assert.Equal("Operator Key", result.Principal.FindFirst(ClaimTypes.Name)?.Value);
+        Assert.Contains(result.Principal.Claims, claim =>
+            claim.Type == DashboardAuthenticationDefaults.ScopeClaimType
+            && claim.Value == GatewayScopes.Admin);
+        Assert.Equal("Bearer mxgw_operator01_super-secret", verifier.LastAuthorizationHeader);
+    }
+
+    [Fact]
+    public async Task AuthenticateAsync_NonAdminKey_ReturnsFailureWithoutRawApiKey()
+    {
+        DashboardAuthenticator authenticator = CreateAuthenticator(new FakeApiKeyVerifier(
+            SuccessWithScopes(GatewayScopes.EventsRead)));
+
+        DashboardAuthenticationResult result = await authenticator.AuthenticateAsync(
+            "mxgw_operator01_super-secret",
+            CancellationToken.None);
+
+        Assert.False(result.Succeeded);
+        Assert.Null(result.Principal);
+        Assert.DoesNotContain("super-secret", result.FailureMessage, StringComparison.Ordinal);
+    }
+
+    [Fact]
+    public async Task AuthenticateAsync_RequireAdminScopeFalse_AllowsAuthenticatedKey()
+    {
+        DashboardAuthenticator authenticator = CreateAuthenticator(
+            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.EventsRead)),
+            requireAdminScope: false);
+
+        DashboardAuthenticationResult result = await authenticator.AuthenticateAsync(
+            "mxgw_operator01_secret",
+            CancellationToken.None);
+
+        Assert.True(result.Succeeded);
+        Assert.NotNull(result.Principal);
+    }
+
+    [Fact]
+    public async Task AuthenticateAsync_InvalidKey_ReturnsGenericFailure()
+    {
+        DashboardAuthenticator authenticator = CreateAuthenticator(new FakeApiKeyVerifier(
+            ApiKeyVerificationResult.Fail(ApiKeyVerificationFailure.SecretMismatch)));
+
+        DashboardAuthenticationResult result = await authenticator.AuthenticateAsync(
+            "mxgw_operator01_super-secret",
+            CancellationToken.None);
+
+        Assert.False(result.Succeeded);
+        Assert.DoesNotContain("super-secret", result.FailureMessage, StringComparison.Ordinal);
+    }
+
+    private static DashboardAuthenticator CreateAuthenticator(
+        IApiKeyVerifier verifier,
+        bool requireAdminScope = true)
+    {
+        return new DashboardAuthenticator(
+            verifier,
+            Options.Create(new GatewayOptions
+            {
+                Dashboard = new DashboardOptions
+                {
+                    RequireAdminScope = requireAdminScope
+                }
+            }));
+    }
+
+    private static ApiKeyVerificationResult SuccessWithScopes(params string[] scopes)
+    {
+        return ApiKeyVerificationResult.Success(new ApiKeyIdentity(
+            KeyId: "operator01",
+            KeyPrefix: "mxgw_operator01",
+            DisplayName: "Operator Key",
+            Scopes: new HashSet<string>(scopes, StringComparer.Ordinal)));
+    }
+
+    private sealed class FakeApiKeyVerifier(ApiKeyVerificationResult result) : IApiKeyVerifier
+    {
+        public string? LastAuthorizationHeader { get; private set; }
+
+        public Task<ApiKeyVerificationResult> VerifyAsync(
+            string? authorizationHeader,
+            CancellationToken cancellationToken)
+        {
+            LastAuthorizationHeader = authorizationHeader;
+
+            return Task.FromResult(result);
+        }
+    }
+}

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

@@ -0,0 +1,91 @@
+using System.Net;
+using System.Security.Claims;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Options;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Dashboard;
+using MxGateway.Server.Security.Authorization;
+
+namespace MxGateway.Tests.Gateway.Dashboard;
+
+public sealed class DashboardAuthorizationHandlerTests
+{
+    [Fact]
+    public async Task HandleAsync_UnauthenticatedRemoteRequest_DoesNotSucceed()
+    {
+        AuthorizationHandlerContext context = await AuthorizeAsync(
+            new ClaimsPrincipal(new ClaimsIdentity()),
+            IPAddress.Parse("10.0.0.5"),
+            allowAnonymousLocalhost: false);
+
+        Assert.False(context.HasSucceeded);
+    }
+
+    [Fact]
+    public async Task HandleAsync_AnonymousLocalhostAllowed_Succeeds()
+    {
+        AuthorizationHandlerContext context = await AuthorizeAsync(
+            new ClaimsPrincipal(new ClaimsIdentity()),
+            IPAddress.Loopback,
+            allowAnonymousLocalhost: true);
+
+        Assert.True(context.HasSucceeded);
+    }
+
+    [Fact]
+    public async Task HandleAsync_AuthenticatedWithoutAdminScope_DoesNotSucceed()
+    {
+        AuthorizationHandlerContext context = await AuthorizeAsync(
+            CreatePrincipal(GatewayScopes.EventsRead),
+            IPAddress.Loopback,
+            allowAnonymousLocalhost: false);
+
+        Assert.False(context.HasSucceeded);
+    }
+
+    [Fact]
+    public async Task HandleAsync_AuthenticatedWithAdminScope_Succeeds()
+    {
+        AuthorizationHandlerContext context = await AuthorizeAsync(
+            CreatePrincipal(GatewayScopes.Admin),
+            IPAddress.Parse("10.0.0.5"),
+            allowAnonymousLocalhost: false);
+
+        Assert.True(context.HasSucceeded);
+    }
+
+    private static async Task<AuthorizationHandlerContext> AuthorizeAsync(
+        ClaimsPrincipal principal,
+        IPAddress remoteAddress,
+        bool allowAnonymousLocalhost)
+    {
+        DashboardAuthorizationRequirement requirement = new();
+        DefaultHttpContext httpContext = new();
+        httpContext.Connection.RemoteIpAddress = remoteAddress;
+        DashboardAuthorizationHandler handler = new(
+            new HttpContextAccessor { HttpContext = httpContext },
+            Options.Create(new GatewayOptions
+            {
+                Dashboard = new DashboardOptions
+                {
+                    AllowAnonymousLocalhost = allowAnonymousLocalhost,
+                    RequireAdminScope = true
+                }
+            }));
+        AuthorizationHandlerContext context = new([requirement], principal, httpContext);
+
+        await handler.HandleAsync(context);
+
+        return context;
+    }
+
+    private static ClaimsPrincipal CreatePrincipal(string scope)
+    {
+        ClaimsIdentity identity = new(
+            [new Claim(DashboardAuthenticationDefaults.ScopeClaimType, scope)],
+            DashboardAuthenticationDefaults.AuthenticationScheme);
+
+        return new ClaimsPrincipal(identity);
+    }
+}

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

@@ -0,0 +1,32 @@
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Authentication.Cookies;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+using MxGateway.Server;
+using MxGateway.Server.Dashboard;
+
+namespace MxGateway.Tests.Gateway.Dashboard;
+
+public sealed class DashboardCookieOptionsTests
+{
+    [Fact]
+    public void Build_ConfiguresSecureDashboardCookie()
+    {
+        WebApplication app = GatewayApplication.Build([]);
+        IOptionsMonitor<CookieAuthenticationOptions> optionsMonitor = app.Services
+            .GetRequiredService<IOptionsMonitor<CookieAuthenticationOptions>>();
+
+        CookieAuthenticationOptions options = optionsMonitor.Get(
+            DashboardAuthenticationDefaults.AuthenticationScheme);
+
+        Assert.Equal(DashboardAuthenticationDefaults.CookieName, options.Cookie.Name);
+        Assert.True(options.Cookie.HttpOnly);
+        Assert.Equal(CookieSecurePolicy.Always, options.Cookie.SecurePolicy);
+        Assert.Equal(SameSiteMode.Strict, options.Cookie.SameSite);
+        Assert.Equal("/", options.Cookie.Path);
+        Assert.Equal("/dashboard/login", options.LoginPath);
+        Assert.Equal("/dashboard/logout", options.LogoutPath);
+        Assert.Equal("/dashboard/denied", options.AccessDeniedPath);
+    }
+}

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

@@ -1,383 +0,0 @@
-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

@@ -184,7 +184,6 @@ public sealed class MxAccessGatewayServiceTests
             identityAccessor ?? new GatewayRequestIdentityAccessor(),
             new MxAccessGrpcRequestValidator(),
             new MxAccessGrpcMapper(),
-            new FakeEventStreamService(sessionManager),
             NullLogger<MxAccessGatewayService>.Instance);
     }
 
@@ -276,11 +275,6 @@ public sealed class MxAccessGatewayServiceTests
 
         public List<WorkerEvent> Events { get; } = [];
 
-        public void RecordReadEventsSessionId(string sessionId)
-        {
-            LastReadEventsSessionId = sessionId;
-        }
-
         public Task<GatewaySession> OpenSessionAsync(
             SessionOpenRequest request,
             string? clientIdentity,
@@ -349,27 +343,6 @@ public sealed class MxAccessGatewayServiceTests
         }
     }
 
-    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";

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

@@ -109,32 +109,6 @@ 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()
     {