MCP OAuth: replica-safe authorization codes via mesh-backed store

memex/Memex.Portal.Shared/Authentication/OAuthCodeNodeType.cs

@@ -0,0 +1,53 @@
+using MeshWeaver.Graph;
+using MeshWeaver.Graph.Configuration;
+using MeshWeaver.Mesh;
+using MeshWeaver.Messaging;
+
+namespace Memex.Portal.Shared.Authentication;
+
+/// <summary>
+/// NodeType definition for <see cref="AuthorizationCode"/> nodes — the pending OAuth
+/// authorization codes <see cref="OAuthCodeStore"/> persists at
+/// <c>Admin/OAuthCode/{hashPrefix}</c> so a /token exchange on ANY portal replica can
+/// redeem a code minted by any other (the in-memory predecessor broke under KEDA
+/// scale-out). Short-lived infrastructure rows: single-use (consumed by delete on
+/// exchange), 5-minute lifetime, System-identity managed — excluded from search,
+/// create menus, and autocomplete.
+/// </summary>
+public static class OAuthCodeNodeType
+{
+    /// <summary>The node-type identifier string for OAuth authorization-code nodes.</summary>
+    public const string NodeType = "OAuthCode";
+
+    /// <summary>
+    /// Registers the OAuthCode node type on the mesh builder: adds the MeshNode type
+    /// definition, excludes it from autocomplete, and registers the
+    /// <see cref="AuthorizationCode"/> content type so code nodes (de)serialize across
+    /// silos and persistence round-trips.
+    /// </summary>
+    public static TBuilder AddOAuthCodeType<TBuilder>(this TBuilder builder) where TBuilder : MeshBuilder
+    {
+        builder.AddMeshNodes(CreateMeshNode());
+        builder.AddAutocompleteExcludedTypes(NodeType);
+        builder.WithMeshType<AuthorizationCode>();
+        return builder;
+    }
+
+    /// <summary>Builds the MeshNode type definition for OAuth authorization codes.</summary>
+    public static MeshNode CreateMeshNode() => new(NodeType)
+    {
+        Name = "OAuth Authorization Code",
+        NodeType = "NodeType",
+        Icon = "/static/NodeTypeIcons/key.svg",
+        IsSatelliteType = false,
+        ExcludeFromContext = System.Collections.Immutable.ImmutableHashSet.Create("search", "create"),
+        Content = new NodeTypeDefinition
+        {
+            Description = "Pending OAuth authorization code (PKCE) awaiting its /token exchange. "
+                          + "Stored under Admin/OAuthCode/{hashPrefix}; single-use and short-lived.",
+        },
+        HubConfiguration = config => config
+            .AddMeshDataSource(source => source
+                .WithContentType<AuthorizationCode>())
+    };
+}

memex/Memex.Portal.Shared/Authentication/OAuthConnectController.cs

@@ -131,16 +131,20 @@ public IActionResult RegisterClient([FromBody] ClientRegistrationRequest? reques
     /// <summary>
     /// OAuth Authorization Endpoint — redirects authenticated users to the client's redirect_uri
     /// with an authorization code. Unauthenticated users are sent to /login first.
+    /// The code is persisted as a mesh node (replica-safe: the /token exchange may land on a
+    /// different pod), so this action bridges the store's IObservable to Task at the HTTP
+    /// boundary — same single-bridge shape as <see cref="ExchangeToken"/>.
     /// </summary>
     [HttpGet("/authorize")]
-    public IActionResult Authorize(
+    public Task<IActionResult> Authorize(
         [FromQuery] string response_type,
         [FromQuery] string client_id,
         [FromQuery] string redirect_uri,
         [FromQuery] string? state,
         [FromQuery] string? scope,
         [FromQuery] string? code_challenge,
-        [FromQuery] string? code_challenge_method)
+        [FromQuery] string? code_challenge_method,
+        CancellationToken ct = default)
     {
         logger.LogInformation(
             "OAuth /authorize: response_type={ResponseType}, client_id={ClientId}, redirect_uri={RedirectUri}, has_state={HasState}, has_pkce={HasPkce}, authenticated={Authenticated}",
@@ -151,13 +155,13 @@ public IActionResult Authorize(
         if (response_type != "code")
         {
             logger.LogWarning("OAuth /authorize rejected: unsupported response_type={ResponseType}", response_type);
-            return BadRequest(new { error = "unsupported_response_type" });
+            return Task.FromResult<IActionResult>(BadRequest(new { error = "unsupported_response_type" }));
         }
 
         if (string.IsNullOrEmpty(client_id) || string.IsNullOrEmpty(redirect_uri))
         {
             logger.LogWarning("OAuth /authorize rejected: missing client_id or redirect_uri");
-            return BadRequest(new { error = "invalid_request", error_description = "client_id and redirect_uri are required" });
+            return Task.FromResult<IActionResult>(BadRequest(new { error = "invalid_request", error_description = "client_id and redirect_uri are required" }));
         }
 
         // If user is not authenticated, redirect to login with return URL
@@ -166,7 +170,7 @@ public IActionResult Authorize(
             var authorizeUrl = $"{Request.Scheme}://{Request.Host}{Request.Path}{Request.QueryString}";
             var loginUrl = $"/login?returnUrl={Uri.EscapeDataString(authorizeUrl)}";
             logger.LogInformation("OAuth /authorize: redirecting unauthenticated caller to {LoginUrl}", loginUrl);
-            return Redirect(loginUrl);
+            return Task.FromResult<IActionResult>(Redirect(loginUrl));
         }
 
         // Extract user identity from cookie claims (email/name are display
@@ -185,7 +189,7 @@ public IActionResult Authorize(
         if (string.IsNullOrEmpty(email))
         {
             logger.LogWarning("OAuth /authorize rejected: authenticated principal has no email/preferred_username claim");
-            return BadRequest(new { error = "invalid_request", error_description = "Unable to determine user identity" });
+            return Task.FromResult<IActionResult>(BadRequest(new { error = "invalid_request", error_description = "Unable to determine user identity" }));
         }
 
         // Refuse to issue a code with an unresolved or email-shaped userId — it
@@ -199,27 +203,43 @@ public IActionResult Authorize(
                 "OAuth /authorize rejected: no resolved mesh identity for {Email} (userId='{UserId}'). "
                 + "Retry after a browser login provisions/loads the User node.",
                 email, userId ?? "(null)");
-            return BadRequest(new { error = "invalid_request", error_description = "Unable to determine user identity" });
+            return Task.FromResult<IActionResult>(BadRequest(new { error = "invalid_request", error_description = "Unable to determine user identity" }));
         }
 
-        // Generate authorization code
-        var code = CodeStore.GenerateCode(
-            userId: userId,
-            userName: name,
-            userEmail: email,
-            clientId: client_id,
-            redirectUri: redirect_uri,
-            codeChallenge: code_challenge,
-            codeChallengeMethod: code_challenge_method);
-
-        logger.LogInformation("Issued OAuth authorization code for user {Email}, client {ClientId}", email, client_id);
+        // Generate + persist the authorization code (mesh node — visible to every
+        // replica). The redirect only fires once the node write committed; a code
+        // whose persistence failed must never reach the client (it could not be
+        // exchanged anywhere), so a store error surfaces as 500 server_error.
+        return CodeStore.GenerateCode(
+                userId: userId,
+                userName: name,
+                userEmail: email,
+                clientId: client_id,
+                redirectUri: redirect_uri,
+                codeChallenge: code_challenge,
+                codeChallengeMethod: code_challenge_method)
+            .Select(code =>
+            {
+                logger.LogInformation("Issued OAuth authorization code for user {Email}, client {ClientId}", email, client_id);
 
-        // Redirect to client with code (and state if provided)
-        var callbackUrl = string.IsNullOrEmpty(state)
-            ? $"{redirect_uri}?code={Uri.EscapeDataString(code)}"
-            : $"{redirect_uri}?code={Uri.EscapeDataString(code)}&state={Uri.EscapeDataString(state)}";
+                // Redirect to client with code (and state if provided)
+                var callbackUrl = string.IsNullOrEmpty(state)
+                    ? $"{redirect_uri}?code={Uri.EscapeDataString(code)}"
+                    : $"{redirect_uri}?code={Uri.EscapeDataString(code)}&state={Uri.EscapeDataString(state)}";
 
-        return Redirect(callbackUrl);
+                return (IActionResult)Redirect(callbackUrl);
+            })
+            .Catch<IActionResult, Exception>(ex =>
+            {
+                logger.LogError(ex,
+                    "OAuth /authorize failed to persist the authorization code for user {Email}, client {ClientId}",
+                    email, client_id);
+                return Observable.Return<IActionResult>(StatusCode(
+                    StatusCodes.Status500InternalServerError,
+                    new { error = "server_error", error_description = "Failed to persist the authorization code" }));
+            })
+            .FirstAsync()
+            .ToTask(ct);
     }
 
     /// <summary>
@@ -247,49 +267,52 @@ public Task<IActionResult> ExchangeToken([FromForm] TokenRequest request, Cancel
             return Task.FromResult<IActionResult>(BadRequest(new { error = "invalid_request" }));
         }
 
-        var entry = CodeStore.ExchangeCode(
-            request.code,
-            request.client_id,
-            request.redirect_uri,
-            request.code_verifier,
-            out var failureReason);
-
-        if (entry == null)
-        {
-            // The reason names the exact failing check (unknown/consumed, expired, client_id,
-            // redirect_uri, PKCE) — the wire response stays a generic invalid_grant per RFC 6749.
-            logger.LogWarning(
-                "OAuth token exchange failed for client {ClientId}: {FailureReason}",
-                request.client_id, failureReason);
-            return Task.FromResult<IActionResult>(BadRequest(new { error = "invalid_grant" }));
-        }
-
-        // Create an mw_ API token via the existing token service. Lifetime
-        // is long-lived because OAuth clients (MCP, CLI tools) typically
-        // can't run interactive re-auth flows — a token that expires in 30
-        // days surprises users who connect once and come back months later.
-        // Refresh-token flow isn't implemented yet; until it is, default to
-        // 1 year. Bump if needed via TokenLifetime below.
-        //
-        // No await: pull IObservable up to the controller's return type.
-        // Single bridge to Task happens at .ToTask(ct) — passing the
-        // request's cancellation token so a client disconnect tears down
-        // the reactive subscription.
-        return TokenService.CreateToken(
-                userId: entry.UserId,
-                userName: entry.UserName,
-                userEmail: entry.UserEmail,
-                label: $"OAuth: {request.client_id}",
-                expiresAt: DateTimeOffset.UtcNow.Add(TokenLifetime))
-            .Select(creation =>
+        // Exchange the code against the mesh-backed store (replica-safe: the code may
+        // have been minted by any pod, and the single-use consume is atomic across
+        // replicas — first delete wins), then mint the mw_ API token. One reactive
+        // chain, single bridge to Task at .ToTask(ct).
+        return CodeStore.ExchangeCode(
+                request.code,
+                request.client_id,
+                request.redirect_uri,
+                request.code_verifier)
+            .SelectMany(exchange =>
             {
-                logger.LogInformation("Issued OAuth access token for user {Email}, client {ClientId}", entry.UserEmail, request.client_id);
-                return (IActionResult)Ok(new
+                if (exchange.Entry is null)
                 {
-                    access_token = creation.RawToken,
-                    token_type = "Bearer",
-                    expires_in = (int)TokenLifetime.TotalSeconds,
-                });
+                    // The reason names the exact failing check (unknown/consumed, lost consume
+                    // race, expired, client_id, redirect_uri, PKCE) — the wire response stays
+                    // a generic invalid_grant per RFC 6749.
+                    logger.LogWarning(
+                        "OAuth token exchange failed for client {ClientId}: {FailureReason}",
+                        request.client_id, exchange.FailureReason);
+                    return Observable.Return<IActionResult>(BadRequest(new { error = "invalid_grant" }));
+                }
+
+                var entry = exchange.Entry;
+
+                // Create an mw_ API token via the existing token service. Lifetime
+                // is long-lived because OAuth clients (MCP, CLI tools) typically
+                // can't run interactive re-auth flows — a token that expires in 30
+                // days surprises users who connect once and come back months later.
+                // Refresh-token flow isn't implemented yet; until it is, default to
+                // 1 year. Bump if needed via TokenLifetime below.
+                return TokenService.CreateToken(
+                        userId: entry.UserId,
+                        userName: entry.UserName,
+                        userEmail: entry.UserEmail,
+                        label: $"OAuth: {request.client_id}",
+                        expiresAt: DateTimeOffset.UtcNow.Add(TokenLifetime))
+                    .Select(creation =>
+                    {
+                        logger.LogInformation("Issued OAuth access token for user {Email}, client {ClientId}", entry.UserEmail, request.client_id);
+                        return (IActionResult)Ok(new
+                        {
+                            access_token = creation.RawToken,
+                            token_type = "Bearer",
+                            expires_in = (int)TokenLifetime.TotalSeconds,
+                        });
+                    });
             })
             .FirstAsync()
             .ToTask(ct);

memex/Memex.Portal.Shared/MemexConfiguration.cs

@@ -652,6 +652,13 @@ public TBuilder ConfigureMemexMesh(IConfiguration configuration, bool isDevelopm
                 // credential nodes. Without this the create throws "NodeType 'ApiCredential' is not
                 // registered" — the OAuth callback's persist step fails and sign-in reports failure.
                 .AddApiCredentialType()
+                // Register the OAuthCode NodeType + AuthorizationCode content type so the
+                // MCP OAuth server (OAuthCodeStore) can persist pending authorization codes
+                // as Admin/OAuthCode/{hashPrefix} mesh nodes — the replica-safe store every
+                // pod shares (the /token exchange may land on a different replica than the
+                // /authorize that minted the code). Without this the create fails with
+                // "NodeType 'OAuthCode' is not registered" and no MCP client can connect.
+                .AddOAuthCodeType()
                 // Seed root-scope Admin AccessAssignments for users listed under
                 // `Auth:GlobalAdmins` so configured admins bypass per-partition
                 // RLS for cross-partition operations (list Spaces, create

src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs

@@ -335,6 +335,26 @@ internal static ISynchronizationStream CreateExternalClient<TReduced, TReference
                         reduced.StreamId),
                 ex =>
                 {
+                    // 🚨 TRANSIENT shutdown reject (ErrorType.ShuttingDown): our SubscribeRequest
+                    // raced the owner hub's DisposeRequest — a recycle / restart in flight, NOT a
+                    // missing owner. The owner's fresh activation re-announces via the change feed
+                    // and the resubscribe latch below re-sends the SubscribeRequest (~2s later in
+                    // the recycle flow), so the ONLY correct reaction is to keep the stream and
+                    // the keep-alive ALIVE and wait for that rehydration — exactly what happened
+                    // under the pre-NACK silent drop. Faulting here killed the latch with the
+                    // stream and wedged every reader of a mid-recycle NodeType to its timeout
+                    // (CI 30003419841, NodeTypeCompileParkTest.RecycleRetry). This is not a
+                    // retry/watchdog: nothing re-posts on a timer — recovery stays event-driven
+                    // (change-feed announce → latch), the same protocol as an owner restart.
+                    if (ex is DeliveryFailureException { Failure.ErrorType: ErrorType.ShuttingDown })
+                    {
+                        logger.LogDebug(
+                            "SubscribeRequest for stream {StreamId} rejected — owner {Owner} is shutting down "
+                            + "(recycle/restart in flight); keeping the stream alive for the resubscribe latch",
+                            reduced.StreamId, owner);
+                        return;
+                    }
+
                     // DeliveryFailure / NotFound — the owner address does not exist. Terminal: fault
                     // subscribers + tear down the keep-alive so we NEVER heartbeat or resubscribe a
                     // non-existent owner (the storm that wedges the session).

src/MeshWeaver.Data/Serialization/SynchronizationStream.cs

@@ -973,6 +973,25 @@ private MessageHubConfiguration ConfigureSynchronizationHub(MessageHubConfigurat
             ).WithHandler<DeliveryFailure>((_, delivery) =>
                 {
                     var failure = delivery.Message;
+                    // 🚨 A TRANSIENT shutdown reject is NOT terminal for a sync stream. It
+                    // means one delivery (typically our SubscribeRequest) raced the owner
+                    // hub's DisposeRequest during a recycle/restart — the address is about
+                    // to reactivate. This stream's own recovery machinery (keep-alive +
+                    // the change-feed resubscribe latch) re-sends the SubscribeRequest
+                    // once the fresh activation announces itself, so the right reaction is
+                    // to RIDE IT OUT. Calling OnError here tore that machinery down with
+                    // the stream: the latch never fired, nothing rehydrated, and every
+                    // reader of a mid-recycle NodeType waited to its timeout
+                    // (CI 30003419841, NodeTypeCompileParkTest.RecycleRetry — the exact
+                    // regression the ShuttingDown ErrorType exists to prevent). All other
+                    // failure kinds (RLS denial, validation, NotFound, …) stay terminal.
+                    if (failure.ErrorType == ErrorType.ShuttingDown)
+                    {
+                        logger.LogDebug(
+                            "Stream {StreamId} received transient shutdown reject: {Message} — keeping the stream alive for the resubscribe latch",
+                            StreamId, failure.Message);
+                        return delivery.Processed();
+                    }
                     logger.LogWarning("Stream {StreamId} received DeliveryFailure: {Message}", StreamId, failure.Message);
                     OnError(new DeliveryFailureException(failure));
                     return delivery.Processed();

src/MeshWeaver.Documentation/Data/WhatsNew/2026-07-23-mcp-connect-reliability.md

@@ -0,0 +1,10 @@
+---
+Name: Reliable MCP sign-in on scaled deployments
+Category: What's New
+Description: Connecting an MCP client (Claude Code, Copilot, IDEs) now works reliably on portals running multiple replicas.
+Icon: Sparkle
+---
+
+# Reliable MCP sign-in on scaled deployments
+
+Connecting an MCP client to the portal could fail with an authorization error when the portal was running more than one replica — the sign-in would only succeed if two consecutive requests happened to reach the same server instance. Authorization codes are now stored in the mesh itself, so any replica can complete a sign-in started on any other. Codes remain single-use and short-lived, and interrupted sign-ins now fail fast with a clear error instead of hanging.