From 5d29c350aa420a7d40c2282722503bb821e86751 Mon Sep 17 00:00:00 2001 From: Bazyli Brzoska Date: Fri, 7 Aug 2026 16:51:58 -0700 Subject: [PATCH 1/3] fix(core): single-flight remote catalog sync Share the complete tool-production operation for concurrent refreshes of the same connection, including upstream discovery and catalog persistence. Explicit refreshes promote an in-flight background refresh so authoritative empty catalogs retain explicit semantics. This prevents simultaneous Desktop tool reads from opening overlapping SQLite transactions, which previously produced 'cannot start a transaction within a transaction' and left MCP integrations degraded despite a usable cached catalog. Preserve MCP connector and discovery failure messages as incomplete-catalog reasons so connection health reports the actionable transport failure instead of the generic 'plugin returned an incomplete tool catalog'. Regression coverage exercises concurrent stale reads and verifies MCP discovery failures reach degraded health. Live verification against the active Gateway configuration completed eight concurrent refreshes with eight HTTP 200 responses and no SQLite collision. --- packages/core/sdk/src/connections.test.ts | 70 ++++++++++++++++++- packages/core/sdk/src/executor.ts | 40 +++++++++-- .../plugins/mcp/src/sdk/catalog-sync.test.ts | 19 +++++ packages/plugins/mcp/src/sdk/plugin.ts | 32 +++++---- 4 files changed, 144 insertions(+), 17 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 7ed046f65..aa133cf35 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Predicate, Result } from "effect"; +import { Deferred, Effect, Fiber, Predicate, Result } from "effect"; import { AuthTemplateSlug, @@ -361,6 +361,74 @@ describe("connections.refresh", () => { }); describe("tool catalog sync safety", () => { + it.effect("single-flights concurrent refreshes of the same stale connection", () => + Effect.scoped( + Effect.gen(function* () { + const refreshStarted = yield* Deferred.make(); + const releaseRefresh = yield* Deferred.make(); + let resolutions = 0; + const guardedPlugin = definePlugin(() => ({ + id: "guarded" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + remoteToolCatalog: true, + resolveTools: () => + Effect.gen(function* () { + resolutions += 1; + if (resolutions > 1) { + yield* Deferred.succeed(refreshStarted, undefined); + yield* Deferred.await(releaseRefresh); + } + return { + tools: [{ name: ToolName.make("deploy"), description: "deploy" }], + }; + }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Vercel", + config: {}, + }), + }), + }))(); + const config = makeTestConfig({ plugins: [guardedPlugin] as const }); + const executor = yield* createExecutor(config); + yield* executor.guarded.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), + set: { tools_synced_at: null }, + }), + ); + + const readsFiber = yield* Effect.forkChild( + Effect.all( + [ + executor.tools.list({ integration: INTEG }), + executor.tools.list({ integration: INTEG }), + ], + { concurrency: "unbounded" }, + ), + ); + yield* Deferred.await(refreshStarted); + yield* Deferred.succeed(releaseRefresh, undefined); + const reads = yield* Fiber.join(readsFiber); + + expect(reads).toHaveLength(2); + expect(resolutions).toBe(2); + }), + ), + ); + it.effect( "background sync preserves a nonzero remote catalog when a plugin returns authoritative empty", () => diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 56cb2e299..43b77c041 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1,4 +1,4 @@ -import { Effect, Inspectable, Layer, Option, Predicate, Schema } from "effect"; +import { Deferred, Effect, Inspectable, Layer, Option, Predicate, Schema } from "effect"; import { FetchHttpClient, type HttpClient } from "effect/unstable/http"; import { fumadb } from "@executor-js/fumadb"; import { memoryAdapter } from "@executor-js/fumadb/adapters/memory"; @@ -2506,10 +2506,10 @@ export const createExecutor = result.incompleteReason ?? "plugin returned an incomplete tool catalog"; - const produceConnectionTools = ( + const produceConnectionToolsUnshared = ( integrationRow: IntegrationRow, ref: ConnectionRef, - mode: "explicit" | "background" = "explicit", + mode: () => "explicit" | "background", ): Effect.Effect => Effect.gen(function* () { const runtime = runtimes.get(integrationRow.plugin_id); @@ -2631,7 +2631,7 @@ export const createExecutor = ; + mode: "explicit" | "background"; + } + const toolProductionInFlight = new Map(); + const produceConnectionTools = ( + integrationRow: IntegrationRow, + ref: ConnectionRef, + requestedMode: "explicit" | "background" = "explicit", + ): Effect.Effect => + Effect.suspend(() => { + const key = `${ref.owner}:${String(ref.integration)}:${String(ref.name)}`; + const existing = toolProductionInFlight.get(key); + if (existing) { + if (requestedMode === "explicit") existing.mode = "explicit"; + return Deferred.await(existing.deferred); + } + + const entry: ToolProductionInFlight = { + deferred: Deferred.makeUnsafe(), + mode: requestedMode, + }; + toolProductionInFlight.set(key, entry); + const run = produceConnectionToolsUnshared(integrationRow, ref, () => entry.mode).pipe( + Effect.exit, + Effect.flatMap((exit) => Deferred.done(entry.deferred, exit)), + Effect.ensuring(Effect.sync(() => void toolProductionInFlight.delete(key))), + ); + return Effect.forkDetach(run).pipe(Effect.andThen(Deferred.await(entry.deferred))); + }); + // ------------------------------------------------------------------ // Connections // ------------------------------------------------------------------ diff --git a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts index 594848571..88c37ab5b 100644 --- a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts +++ b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts @@ -163,6 +163,25 @@ describe("MCP tool-catalog sync (end-to-end)", () => { expect(server.sessionCount()).toBe(sessionsAfterFirstList); }), ); + + it.effect("preserves the MCP discovery failure in degraded connection health", () => + Effect.gen(function* () { + const server = yield* serveTestHttpApp(() => + Effect.succeed(HttpServerResponse.text("gateway unavailable", { status: 503 })), + ); + const executor = yield* makeCatalogTestExecutor(server.url("/mcp")); + const connection = yield* executor.connections.get({ + owner: "org", + integration: INTEG, + name: CONNECTION, + }); + + expect(connection?.lastHealth).toMatchObject({ + status: "degraded", + detail: expect.stringContaining("Failed connecting to MCP server"), + }); + }), + ); }); // --------------------------------------------------------------------------- diff --git a/packages/plugins/mcp/src/sdk/plugin.ts b/packages/plugins/mcp/src/sdk/plugin.ts index e3b7a6857..234d77ec1 100644 --- a/packages/plugins/mcp/src/sdk/plugin.ts +++ b/packages/plugins/mcp/src/sdk/plugin.ts @@ -1233,20 +1233,28 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { Effect.result, ); - const manifest = Result.isSuccess(built) - ? yield* discoverTools(built.success).pipe( - Effect.map((m) => ({ ok: true as const, manifest: m })), - Effect.catch(() => Effect.succeed({ ok: false as const, manifest: null })), - Effect.withSpan("mcp.plugin.discover_tools", { - attributes: { "mcp.connection.name": String(connection.name) }, - }), - ) - : { ok: false as const, manifest: null }; + if (Result.isFailure(built)) { + return { + tools: [] as readonly ToolDef[], + incomplete: true, + incompleteReason: built.failure.message, + }; + } - if (!manifest.ok || !manifest.manifest) { - return { tools: [] as readonly ToolDef[], incomplete: true }; + const discovered = yield* discoverTools(built.success).pipe( + Effect.result, + Effect.withSpan("mcp.plugin.discover_tools", { + attributes: { "mcp.connection.name": String(connection.name) }, + }), + ); + if (Result.isFailure(discovered)) { + return { + tools: [] as readonly ToolDef[], + incomplete: true, + incompleteReason: discovered.failure.message, + }; } - return { tools: manifest.manifest.tools.map(toToolDef) }; + return { tools: discovered.success.tools.map(toToolDef) }; }).pipe( Effect.withSpan("mcp.plugin.resolve_tools", { attributes: { "mcp.connection.name": String(connection.name) }, From 5d10a6e39697f47fabaf6a8c3fb3d1a314e4d973 Mon Sep 17 00:00:00 2001 From: Bazyli Brzoska Date: Fri, 7 Aug 2026 16:59:50 -0700 Subject: [PATCH 2/3] fix(cli): preserve custom CA trust in services Propagate NODE_EXTRA_CA_CERTS, SSL_CERT_FILE, and SSL_CERT_DIR from the installing shell into the supervised daemon environment generated for launchd, systemd, and Windows Task Scheduler. Service managers intentionally start Executor with a minimal environment. On machines whose HTTPS trust depends on an enterprise or intercepting CA bundle, dropping these variables made MCP and other HTTPS integrations fail only in the background service while identical calls from the CLI succeeded. MCP auto transport then obscured the TLS failure behind its final SSE fallback error. Add regression coverage at the shared service-environment boundary to ensure all supported CA path variables survive unit generation without embedding certificate contents or other credentials. Reproduced with an isolated compiled daemon: the Gateway succeeded with the shell CA environment and degraded when both CA variables were removed. After rebuilding and reinstalling the LaunchAgent, the active Gateway cleared its degraded health and eight concurrent refreshes returned eight HTTP 200 responses. --- apps/cli/src/service.test.ts | 30 ++++++++++++++++++++++++++++++ apps/cli/src/service.ts | 7 +++++++ 2 files changed, 37 insertions(+) diff --git a/apps/cli/src/service.test.ts b/apps/cli/src/service.test.ts index 4da92cc35..07dcfab19 100644 --- a/apps/cli/src/service.test.ts +++ b/apps/cli/src/service.test.ts @@ -185,6 +185,36 @@ describe("service unit generation", () => { expect(wrapper).not.toContain("EXECUTOR_AUTH_PASSWORD"); }); + it("preserves custom TLS trust paths in the supervised environment", () => { + const previousNodeExtraCaCerts = process.env.NODE_EXTRA_CA_CERTS; + const previousSslCertFile = process.env.SSL_CERT_FILE; + const previousSslCertDir = process.env.SSL_CERT_DIR; + process.env.NODE_EXTRA_CA_CERTS = "/Users/x/.certs/node-extra.pem"; + process.env.SSL_CERT_FILE = "/Users/x/.certs/combined.pem"; + process.env.SSL_CERT_DIR = "/Users/x/.certs"; + + const wrapper = generateWindowsDaemonWrapper( + { + executablePath: "C:\\Program Files\\Executor\\executor.exe", + port: 4789, + version: "1.5.10", + }, + "C:\\Users\\x\\.executor", + "C:\\Users\\x\\.executor\\logs", + ); + + if (previousNodeExtraCaCerts === undefined) delete process.env.NODE_EXTRA_CA_CERTS; + else process.env.NODE_EXTRA_CA_CERTS = previousNodeExtraCaCerts; + if (previousSslCertFile === undefined) delete process.env.SSL_CERT_FILE; + else process.env.SSL_CERT_FILE = previousSslCertFile; + if (previousSslCertDir === undefined) delete process.env.SSL_CERT_DIR; + else process.env.SSL_CERT_DIR = previousSslCertDir; + + expect(wrapper).toContain('set "NODE_EXTRA_CA_CERTS=/Users/x/.certs/node-extra.pem"'); + expect(wrapper).toContain('set "SSL_CERT_FILE=/Users/x/.certs/combined.pem"'); + expect(wrapper).toContain('set "SSL_CERT_DIR=/Users/x/.certs"'); + }); + it("sanitizes cmd.exe metacharacters in baked env values (cmdSetValue)", () => { // A `"` in PATH would close the `set "PATH=..."` quote early and let a // `& cmd &` fragment run at boot as the user; strip it (illegal in a path diff --git a/apps/cli/src/service.ts b/apps/cli/src/service.ts index ade5757fd..ffaf56726 100644 --- a/apps/cli/src/service.ts +++ b/apps/cli/src/service.ts @@ -181,6 +181,13 @@ const serviceEnvironment = ( // or an opted-out install would silently re-enable analytics under launchd. "DO_NOT_TRACK", "EXECUTOR_DISABLE_ANALYTICS", + // Preserve the caller's TLS trust configuration. Corporate/intercepting + // CAs commonly live outside the OS trust store; dropping these paths in a + // minimal launchd/systemd environment makes HTTPS integrations fail only + // after service installation while the same CLI call succeeds. + "NODE_EXTRA_CA_CERTS", + "SSL_CERT_FILE", + "SSL_CERT_DIR", ] as const; const passThrough = Object.fromEntries( passThroughKeys.flatMap((key) => { From d95c2dadad1a2f2b1d7637c40ec7991f471c6cc8 Mon Sep 17 00:00:00 2001 From: Bazyli Brzoska Date: Tue, 11 Aug 2026 15:59:44 -0700 Subject: [PATCH 3/3] fix(mcp): classify remote connection failures Normalize Effect HTTP transport failures into safe structural categories for TLS verification, DNS resolution, timeouts, refused connections, generic network failures, HTTP failures, and MCP protocol incompatibility. The auto transport previously retried every non-auth failure through legacy SSE. When the first Streamable HTTP attempt failed below the protocol layer, the redundant SSE attempt failed for the same reason and replaced the useful primary context with the confusing final message 'Failed connecting via sse'. Retry SSE only when Streamable HTTP reached the endpoint and received evidence of protocol incompatibility. Preserve both attempt summaries when a legitimate fallback also fails, while keeping raw runtime causes internal and excluding upstream response details from customer-facing health messages. Keep connection-handshake numeric HTTP codes separate from JSON-RPC invocation error codes so protocol errors cannot be misclassified as HTTP status failures. Regression coverage verifies that TLS failures produce actionable CA-trust guidance without retrying SSE or leaking internal details, and that an HTTP 405 performs the fallback and reports both attempts. The MCP SDK suite passes 112 tests with 29 skipped; package and workspace typechecks pass. --- .../plugins/mcp/src/sdk/connection.test.ts | 85 +++++++++++ packages/plugins/mcp/src/sdk/connection.ts | 138 ++++++++++++++++-- packages/plugins/mcp/src/sdk/errors.ts | 15 ++ packages/plugins/mcp/src/sdk/http-status.ts | 22 +++ 4 files changed, 251 insertions(+), 9 deletions(-) create mode 100644 packages/plugins/mcp/src/sdk/connection.test.ts diff --git a/packages/plugins/mcp/src/sdk/connection.test.ts b/packages/plugins/mcp/src/sdk/connection.test.ts new file mode 100644 index 000000000..2c20c4ee2 --- /dev/null +++ b/packages/plugins/mcp/src/sdk/connection.test.ts @@ -0,0 +1,85 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Layer } from "effect"; +import { + HttpClient, + HttpClientError, + HttpClientRequest, + HttpClientResponse, +} from "effect/unstable/http"; + +import { createMcpConnector } from "./connection"; + +const endpoint = "https://internal.example/mcp"; + +describe("MCP remote transport failures", () => { + it.effect("surfaces TLS verification failures without retrying SSE", () => + Effect.gen(function* () { + const requests: string[] = []; + const httpClientLayer = Layer.succeed(HttpClient.HttpClient)( + HttpClient.make((request: HttpClientRequest.HttpClientRequest) => { + requests.push(request.url); + return Effect.fail( + new HttpClientError.HttpClientError({ + reason: new HttpClientError.TransportError({ + request, + cause: { + code: "SELF_SIGNED_CERT_IN_CHAIN", + message: "do-not-leak: internal certificate detail", + }, + }), + }), + ); + }), + ); + + const failure = yield* createMcpConnector({ + transport: "remote", + endpoint, + remoteTransport: "auto", + httpClientLayer, + }).pipe(Effect.flip); + + expect(failure).toMatchObject({ + _tag: "McpConnectionError", + transport: "streamable-http", + failureKind: "tls", + message: + "MCP HTTPS connection failed: TLS certificate verification failed. Check the server certificate and Executor's CA trust configuration.", + }); + expect(failure.message).not.toContain("do-not-leak"); + expect(requests).toEqual([endpoint]); + }), + ); + + it.effect("reports both attempts when a protocol mismatch falls back to SSE", () => + Effect.gen(function* () { + const requests: string[] = []; + const httpClientLayer = Layer.succeed(HttpClient.HttpClient)( + HttpClient.make((request: HttpClientRequest.HttpClientRequest) => { + requests.push(`${request.method} ${request.url}`); + return Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response("unsupported MCP transport", { status: 405 }), + ), + ); + }), + ); + + const failure = yield* createMcpConnector({ + transport: "remote", + endpoint, + remoteTransport: "auto", + httpClientLayer, + }).pipe(Effect.flip); + + expect(failure).toMatchObject({ + _tag: "McpConnectionError", + transport: "auto", + failureKind: "protocol", + message: "MCP auto transport failed. Streamable HTTP: HTTP 405. SSE fallback: HTTP 405.", + }); + expect(requests.length).toBeGreaterThanOrEqual(2); + }), + ); +}); diff --git a/packages/plugins/mcp/src/sdk/connection.ts b/packages/plugins/mcp/src/sdk/connection.ts index 82a98cd0d..b3872adca 100644 --- a/packages/plugins/mcp/src/sdk/connection.ts +++ b/packages/plugins/mcp/src/sdk/connection.ts @@ -4,8 +4,8 @@ import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import type { FetchLike } from "@modelcontextprotocol/sdk/shared/transport.js"; import { CfWorkerJsonSchemaValidator } from "@modelcontextprotocol/sdk/validation/cfworker"; -import { Effect, Layer, Predicate, Stream } from "effect"; -import { HttpClient, HttpClientRequest } from "effect/unstable/http"; +import { Effect, Layer, Option, Predicate, Schema, Stream } from "effect"; +import { HttpClient, HttpClientError, HttpClientRequest } from "effect/unstable/http"; // NOTE: `StdioClientTransport` is NOT imported eagerly. The upstream module // (`@modelcontextprotocol/sdk/client/stdio.js`) touches `node:child_process` @@ -18,10 +18,11 @@ import { HttpClient, HttpClientRequest } from "effect/unstable/http"; import type { McpRemoteIntegrationConfig, McpStdioIntegrationConfig } from "./types"; import { McpConnectionError, + McpConnectionFailureKind, McpInsufficientScopeError, McpOAuthReauthorizationRequired, } from "./errors"; -import { httpStatusFromCause } from "./http-status"; +import { connectionHttpStatusFromCause, isStreamableHttpProtocolError } from "./http-status"; import { detectInsufficientScope } from "@executor-js/sdk/core"; // --------------------------------------------------------------------------- @@ -91,6 +92,84 @@ const headersFrom = (headers: HeadersInit | undefined): Headers => const recordFromHeaders = (headers: Headers): Record => Object.fromEntries(headers.entries()); +const ExternalTransportCause = Schema.Struct({ + code: Schema.optional(Schema.String), + cause: Schema.optional(Schema.Unknown), +}); +const decodeExternalTransportCause = Schema.decodeUnknownOption(ExternalTransportCause); + +const TLS_ERROR_CODES = new Set([ + "CERT_HAS_EXPIRED", + "DEPTH_ZERO_SELF_SIGNED_CERT", + "ERR_TLS_CERT_ALTNAME_INVALID", + "SELF_SIGNED_CERT_IN_CHAIN", + "UNABLE_TO_GET_ISSUER_CERT_LOCALLY", + "UNABLE_TO_VERIFY_LEAF_SIGNATURE", +]); +const DNS_ERROR_CODES = new Set(["EAI_AGAIN", "ENOTFOUND"]); +const TIMEOUT_ERROR_CODES = new Set(["ETIMEDOUT", "UND_ERR_CONNECT_TIMEOUT"]); +const PROTOCOL_HTTP_STATUSES = new Set([400, 404, 405, 406, 415, 422, 501]); + +const CONNECTION_FAILURE_MESSAGES: Record = { + tls: "MCP HTTPS connection failed: TLS certificate verification failed. Check the server certificate and Executor's CA trust configuration.", + dns: "MCP connection failed: the server hostname could not be resolved.", + timeout: "MCP connection failed: the server did not respond before the connection timed out.", + connection_refused: "MCP connection failed: the server refused the connection.", + network: "MCP connection failed before transport negotiation completed.", + http: "MCP server rejected the HTTP connection.", + protocol: "MCP server does not support the requested transport.", +}; + +const CONNECTION_ATTEMPT_SUMMARIES: Partial> = { + tls: "TLS certificate verification failed", + dns: "hostname resolution failed", + timeout: "connection timed out", + connection_refused: "connection refused", + protocol: "unsupported protocol response", +}; + +class McpHttpTransportError extends Schema.TaggedErrorClass()( + "McpHttpTransportError", + { + failureKind: McpConnectionFailureKind, + cause: Schema.Defect, + }, +) {} +const decodeMcpHttpTransportError = Schema.decodeUnknownOption(McpHttpTransportError); + +const externalTransportCodes = (cause: unknown): ReadonlySet => { + const codes = new Set(); + let current: unknown = cause; + for (let depth = 0; depth < 8; depth += 1) { + const decoded = decodeExternalTransportCause(current); + if (Option.isNone(decoded)) break; + if (decoded.value.code !== undefined) codes.add(decoded.value.code); + if (decoded.value.cause === undefined) break; + current = decoded.value.cause; + } + return codes; +}; + +const classifyHttpClientFailure = ( + failure: HttpClientError.HttpClientError, +): McpConnectionFailureKind => { + if (!Predicate.isTagged(failure.reason, "TransportError")) return "network"; + const codes = externalTransportCodes(failure.reason.cause); + if ([...codes].some((code) => TLS_ERROR_CODES.has(code))) return "tls"; + if ([...codes].some((code) => DNS_ERROR_CODES.has(code))) return "dns"; + if ([...codes].some((code) => TIMEOUT_ERROR_CODES.has(code))) return "timeout"; + if (codes.has("ECONNREFUSED")) return "connection_refused"; + return "network"; +}; + +const normalizeHttpClientFailure = ( + failure: HttpClientError.HttpClientError, +): McpHttpTransportError => + new McpHttpTransportError({ + failureKind: classifyHttpClientFailure(failure), + cause: failure, + }); + const applyBody = async ( request: HttpClientRequest.HttpClientRequest, headers: Headers, @@ -147,7 +226,7 @@ const fetchFromHttpClientLayer = ( status: response.status, headers: responseHeaders, }); - }).pipe(Effect.provide(httpClientLayer)); + }).pipe(Effect.mapError(normalizeHttpClientFailure), Effect.provide(httpClientLayer)); // A 403 carrying an RFC 6750 insufficient_scope challenge is intercepted // HERE, below the SDK: with an authProvider the SDK would consume the // challenge and re-run auth ("upscoping"), which our static-token @@ -233,17 +312,50 @@ const connectionFailure = ( insufficientScope: true, }); } + const httpTransportError = decodeMcpHttpTransportError(cause); + if (Option.isSome(httpTransportError)) { + return new McpConnectionError({ + transport, + message: CONNECTION_FAILURE_MESSAGES[httpTransportError.value.failureKind], + failureKind: httpTransportError.value.failureKind, + }); + } // Carry the handshake HTTP status structurally (and in the message for // humans) so the liveness health check can classify a rejected credential // as expired rather than a generic connection failure. - const status = httpStatusFromCause(cause); + const status = connectionHttpStatusFromCause(cause); + const failureKind: McpConnectionFailureKind = + isStreamableHttpProtocolError(cause) || + (status !== undefined && PROTOCOL_HTTP_STATUSES.has(status)) + ? "protocol" + : status === undefined + ? "network" + : "http"; return new McpConnectionError({ transport, message: status === undefined ? message : `${message} (HTTP ${status})`, + failureKind, ...(status === undefined ? {} : { httpStatus: status }), }); }; +const connectionAttemptSummary = (failure: McpConnectionError): string => { + if (failure.httpStatus !== undefined) return `HTTP ${failure.httpStatus}`; + return failure.failureKind === undefined + ? "connection failed" + : (CONNECTION_ATTEMPT_SUMMARIES[failure.failureKind] ?? "connection failed"); +}; + +const autoTransportFailure = ( + streamableHttp: McpConnectionError, + sse: McpConnectionError, +): McpConnectionError => + new McpConnectionError({ + transport: "auto", + failureKind: "protocol", + message: `MCP auto transport failed. Streamable HTTP: ${connectionAttemptSummary(streamableHttp)}. SSE fallback: ${connectionAttemptSummary(sse)}.`, + }); + const connectClient = (input: { transport: string; createTransport: () => Parameters[0]; @@ -344,10 +456,18 @@ export const createMcpConnector = (input: ConnectorInput): McpConnector => { // error), which used to misclassify an expired token as a generic // connection failure. Propagate it as-is instead. return connectStreamableHttp.pipe( - Effect.catch((error) => { - if (Predicate.isTagged(error, "McpOAuthReauthorizationRequired")) return Effect.fail(error); - if (error.httpStatus === 401 || error.httpStatus === 403) return Effect.fail(error); - return connectSse; + Effect.catchTags({ + McpOAuthReauthorizationRequired: Effect.fail, + McpConnectionError: (error) => { + if (error.httpStatus === 401 || error.httpStatus === 403) return Effect.fail(error); + if (error.failureKind !== "protocol") return Effect.fail(error); + return connectSse.pipe( + Effect.catchTags({ + McpOAuthReauthorizationRequired: Effect.fail, + McpConnectionError: (sseError) => Effect.fail(autoTransportFailure(error, sseError)), + }), + ); + }, }), ); }; diff --git a/packages/plugins/mcp/src/sdk/errors.ts b/packages/plugins/mcp/src/sdk/errors.ts index e3bba2faf..a56a2c63b 100644 --- a/packages/plugins/mcp/src/sdk/errors.ts +++ b/packages/plugins/mcp/src/sdk/errors.ts @@ -3,11 +3,26 @@ import { Data, Schema } from "effect"; +export const McpConnectionFailureKind = Schema.Literals([ + "tls", + "dns", + "timeout", + "connection_refused", + "network", + "http", + "protocol", +]); +export type McpConnectionFailureKind = typeof McpConnectionFailureKind.Type; + export class McpConnectionError extends Schema.TaggedErrorClass()( "McpConnectionError", { transport: Schema.String, message: Schema.String, + /** Safe, structural classification of the connection failure. Auto + * transport uses this to retry only protocol incompatibilities; callers + * can render actionable copy without parsing an external error message. */ + failureKind: Schema.optional(McpConnectionFailureKind), /** HTTP status the handshake observed (e.g. 401 on an auth wall), when the * transport surfaced one. Structural, so the liveness classifier and the * auto-transport fallback never string-match the message. */ diff --git a/packages/plugins/mcp/src/sdk/http-status.ts b/packages/plugins/mcp/src/sdk/http-status.ts index a4442f8d2..a892da05a 100644 --- a/packages/plugins/mcp/src/sdk/http-status.ts +++ b/packages/plugins/mcp/src/sdk/http-status.ts @@ -13,6 +13,8 @@ import { StreamableHTTPError } from "@modelcontextprotocol/sdk/client/streamable const SsePostErrorCause = Schema.Struct({ message: Schema.String }); const decodeSsePostErrorCause = Schema.decodeUnknownOption(SsePostErrorCause); +const NumericHttpCodeCause = Schema.Struct({ code: Schema.Number }); +const decodeNumericHttpCodeCause = Schema.decodeUnknownOption(NumericHttpCodeCause); // Matches the SDK's SSEClientTransport POST-failure message (sse.js); re-verify // on SDK bumps. A format drift just yields undefined (generic error, no crash). @@ -33,9 +35,29 @@ const statusFromStreamableHttpError = (cause: unknown): number | undefined => { return code !== undefined && code >= 100 && code <= 599 ? code : undefined; }; +const statusFromNumericHttpCode = (cause: unknown): number | undefined => + Option.match(decodeNumericHttpCodeCause(cause), { + onNone: () => undefined, + onSome: ({ code }) => (code >= 100 && code <= 599 ? code : undefined), + }); + export const httpStatusFromCause = (cause: unknown): number | undefined => statusFromStreamableHttpError(cause) ?? statusFromSsePostError(cause); +/** Connection handshakes may receive the SDK's SSE error, whose numeric code + * is an HTTP status. Keep this connection-only: JSON-RPC invocation errors + * also have numeric `code` fields which are not HTTP statuses. */ +export const connectionHttpStatusFromCause = (cause: unknown): number | undefined => + httpStatusFromCause(cause) ?? statusFromNumericHttpCode(cause); + +/** The SDK uses code -1 when Streamable HTTP reached the endpoint but its + * response did not implement the protocol (for example an unexpected content + * type). This is transport incompatibility, not a network outage. */ +export const isStreamableHttpProtocolError = (cause: unknown): boolean => { + // oxlint-disable-next-line executor/no-instanceof-tagged-error -- boundary: MCP SDK exposes this protocol sentinel only on its Error subclass + return cause instanceof StreamableHTTPError && cause.code === -1; +}; + // The SDK embeds the upstream response text in the transport error message // ("Error POSTing to endpoint: "), which is the only place a 403's body // survives for connections without an authProvider. For OAuth connections the