Skip to content

Commit ebdfb44

Browse files
Fix MCP catalog refresh races and transport diagnostics (#1563)
* 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. * 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. * 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. * fix(mcp): retain transport failures through negotiation The v2 MCP client wraps fetch failures from automatic protocol negotiation in an SdkError data.cause chain. Traverse that chain before classifying the connection error, so TLS, DNS, timeout, and refused-connection failures retain their actionable category instead of collapsing into a generic network failure.\n\nThe existing remote-connector regression test executes the negotiated connection path and verifies that a self-signed certificate remains a TLS failure without attempting an SSE fallback. --------- Co-authored-by: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com>
1 parent dbb58f4 commit ebdfb44

10 files changed

Lines changed: 454 additions & 26 deletions

File tree

‎apps/cli/src/service.test.ts‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,36 @@ describe("service unit generation", () => {
185185
expect(wrapper).not.toContain("EXECUTOR_AUTH_PASSWORD");
186186
});
187187

188+
it("preserves custom TLS trust paths in the supervised environment", () => {
189+
const previousNodeExtraCaCerts = process.env.NODE_EXTRA_CA_CERTS;
190+
const previousSslCertFile = process.env.SSL_CERT_FILE;
191+
const previousSslCertDir = process.env.SSL_CERT_DIR;
192+
process.env.NODE_EXTRA_CA_CERTS = "/Users/x/.certs/node-extra.pem";
193+
process.env.SSL_CERT_FILE = "/Users/x/.certs/combined.pem";
194+
process.env.SSL_CERT_DIR = "/Users/x/.certs";
195+
196+
const wrapper = generateWindowsDaemonWrapper(
197+
{
198+
executablePath: "C:\\Program Files\\Executor\\executor.exe",
199+
port: 4789,
200+
version: "1.5.10",
201+
},
202+
"C:\\Users\\x\\.executor",
203+
"C:\\Users\\x\\.executor\\logs",
204+
);
205+
206+
if (previousNodeExtraCaCerts === undefined) delete process.env.NODE_EXTRA_CA_CERTS;
207+
else process.env.NODE_EXTRA_CA_CERTS = previousNodeExtraCaCerts;
208+
if (previousSslCertFile === undefined) delete process.env.SSL_CERT_FILE;
209+
else process.env.SSL_CERT_FILE = previousSslCertFile;
210+
if (previousSslCertDir === undefined) delete process.env.SSL_CERT_DIR;
211+
else process.env.SSL_CERT_DIR = previousSslCertDir;
212+
213+
expect(wrapper).toContain('set "NODE_EXTRA_CA_CERTS=/Users/x/.certs/node-extra.pem"');
214+
expect(wrapper).toContain('set "SSL_CERT_FILE=/Users/x/.certs/combined.pem"');
215+
expect(wrapper).toContain('set "SSL_CERT_DIR=/Users/x/.certs"');
216+
});
217+
188218
it("sanitizes cmd.exe metacharacters in baked env values (cmdSetValue)", () => {
189219
// A `"` in PATH would close the `set "PATH=..."` quote early and let a
190220
// `& cmd &` fragment run at boot as the user; strip it (illegal in a path

‎apps/cli/src/service.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -181,6 +181,13 @@ const serviceEnvironment = (
181181
// or an opted-out install would silently re-enable analytics under launchd.
182182
"DO_NOT_TRACK",
183183
"EXECUTOR_DISABLE_ANALYTICS",
184+
// Preserve the caller's TLS trust configuration. Corporate/intercepting
185+
// CAs commonly live outside the OS trust store; dropping these paths in a
186+
// minimal launchd/systemd environment makes HTTPS integrations fail only
187+
// after service installation while the same CLI call succeeds.
188+
"NODE_EXTRA_CA_CERTS",
189+
"SSL_CERT_FILE",
190+
"SSL_CERT_DIR",
184191
] as const;
185192
const passThrough = Object.fromEntries(
186193
passThroughKeys.flatMap((key) => {

‎packages/core/sdk/src/connections.test.ts‎

Lines changed: 69 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { describe, expect, it } from "@effect/vitest";
2-
import { Effect, Predicate, Result, Schema } from "effect";
2+
import { Deferred, Effect, Fiber, Predicate, Result, Schema } from "effect";
33

44
import {
55
AuthTemplateSlug,
@@ -419,6 +419,74 @@ describe("connections.refresh", () => {
419419
});
420420

421421
describe("tool catalog sync safety", () => {
422+
it.effect("single-flights concurrent refreshes of the same stale connection", () =>
423+
Effect.scoped(
424+
Effect.gen(function* () {
425+
const refreshStarted = yield* Deferred.make<void>();
426+
const releaseRefresh = yield* Deferred.make<void>();
427+
let resolutions = 0;
428+
const guardedPlugin = definePlugin(() => ({
429+
id: "guarded" as const,
430+
credentialProviders: [memoryProvider()],
431+
storage: () => ({}),
432+
remoteToolCatalog: true,
433+
resolveTools: () =>
434+
Effect.gen(function* () {
435+
resolutions += 1;
436+
if (resolutions > 1) {
437+
yield* Deferred.succeed(refreshStarted, undefined);
438+
yield* Deferred.await(releaseRefresh);
439+
}
440+
return {
441+
tools: [{ name: ToolName.make("deploy"), description: "deploy" }],
442+
};
443+
}),
444+
invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }),
445+
extension: (ctx) => ({
446+
seed: () =>
447+
ctx.core.integrations.register({
448+
slug: INTEG,
449+
description: "Vercel",
450+
config: {},
451+
}),
452+
}),
453+
}))();
454+
const config = makeTestConfig({ plugins: [guardedPlugin] as const });
455+
const executor = yield* createExecutor(config);
456+
yield* executor.guarded.seed();
457+
yield* executor.connections.create({
458+
owner: "org",
459+
name: ConnectionName.make("main"),
460+
integration: INTEG,
461+
template: TEMPLATE,
462+
value: "secret-token",
463+
});
464+
yield* Effect.promise(() =>
465+
config.db.updateMany("connection", {
466+
where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")),
467+
set: { tools_synced_at: null },
468+
}),
469+
);
470+
471+
const readsFiber = yield* Effect.forkChild(
472+
Effect.all(
473+
[
474+
executor.tools.list({ integration: INTEG }),
475+
executor.tools.list({ integration: INTEG }),
476+
],
477+
{ concurrency: "unbounded" },
478+
),
479+
);
480+
yield* Deferred.await(refreshStarted);
481+
yield* Deferred.succeed(releaseRefresh, undefined);
482+
const reads = yield* Fiber.join(readsFiber);
483+
484+
expect(reads).toHaveLength(2);
485+
expect(resolutions).toBe(2);
486+
}),
487+
),
488+
);
489+
422490
it.effect(
423491
"background sync preserves a nonzero remote catalog when a plugin returns authoritative empty",
424492
() =>

‎packages/core/sdk/src/executor.ts‎

Lines changed: 36 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { Duration, Effect, Inspectable, Layer, Option, Predicate, Schema } from "effect";
1+
import { Deferred, Duration, Effect, Inspectable, Layer, Option, Predicate, Schema } from "effect";
22
import { FetchHttpClient, type HttpClient } from "effect/unstable/http";
33
import { fumadb } from "@executor-js/fumadb";
44
import { memoryAdapter } from "@executor-js/fumadb/adapters/memory";
@@ -2874,10 +2874,10 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
28742874
const syncHealthReason = (result: ResolveToolsResult): string =>
28752875
result.incompleteReason ?? "plugin returned an incomplete tool catalog";
28762876

2877-
const produceConnectionTools = (
2877+
const produceConnectionToolsUnshared = (
28782878
integrationRow: IntegrationRow,
28792879
ref: ConnectionRef,
2880-
mode: "explicit" | "background" = "explicit",
2880+
mode: () => "explicit" | "background",
28812881
): Effect.Effect<readonly Tool[], IntegrationNotFoundError | StorageFailure> =>
28822882
Effect.gen(function* () {
28832883
const runtime = runtimes.get(integrationRow.plugin_id);
@@ -2999,7 +2999,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
29992999
}
30003000

30013001
if (
3002-
mode === "background" &&
3002+
mode() === "background" &&
30033003
runtime.plugin.remoteToolCatalog === true &&
30043004
result.tools.length === 0
30053005
) {
@@ -3079,6 +3079,38 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
30793079
);
30803080
});
30813081

3082+
type ToolProductionError = IntegrationNotFoundError | StorageFailure;
3083+
interface ToolProductionInFlight {
3084+
readonly deferred: Deferred.Deferred<readonly Tool[], ToolProductionError>;
3085+
mode: "explicit" | "background";
3086+
}
3087+
const toolProductionInFlight = new Map<string, ToolProductionInFlight>();
3088+
const produceConnectionTools = (
3089+
integrationRow: IntegrationRow,
3090+
ref: ConnectionRef,
3091+
requestedMode: "explicit" | "background" = "explicit",
3092+
): Effect.Effect<readonly Tool[], ToolProductionError> =>
3093+
Effect.suspend(() => {
3094+
const key = `${ref.owner}:${String(ref.integration)}:${String(ref.name)}`;
3095+
const existing = toolProductionInFlight.get(key);
3096+
if (existing) {
3097+
if (requestedMode === "explicit") existing.mode = "explicit";
3098+
return Deferred.await(existing.deferred);
3099+
}
3100+
3101+
const entry: ToolProductionInFlight = {
3102+
deferred: Deferred.makeUnsafe<readonly Tool[], ToolProductionError>(),
3103+
mode: requestedMode,
3104+
};
3105+
toolProductionInFlight.set(key, entry);
3106+
const run = produceConnectionToolsUnshared(integrationRow, ref, () => entry.mode).pipe(
3107+
Effect.exit,
3108+
Effect.flatMap((exit) => Deferred.done(entry.deferred, exit)),
3109+
Effect.ensuring(Effect.sync(() => void toolProductionInFlight.delete(key))),
3110+
);
3111+
return Effect.forkDetach(run).pipe(Effect.andThen(Deferred.await(entry.deferred)));
3112+
});
3113+
30823114
// ------------------------------------------------------------------
30833115
// Connections
30843116
// ------------------------------------------------------------------

‎packages/plugins/mcp/src/sdk/catalog-sync.test.ts‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,25 @@ describe("MCP tool-catalog sync (end-to-end)", () => {
163163
expect(server.sessionCount()).toBe(sessionsAfterFirstList);
164164
}),
165165
);
166+
167+
it.effect("preserves the MCP discovery failure in degraded connection health", () =>
168+
Effect.gen(function* () {
169+
const server = yield* serveTestHttpApp(() =>
170+
Effect.succeed(HttpServerResponse.text("gateway unavailable", { status: 503 })),
171+
);
172+
const executor = yield* makeCatalogTestExecutor(server.url("/mcp"));
173+
const connection = yield* executor.connections.get({
174+
owner: "org",
175+
integration: INTEG,
176+
name: CONNECTION,
177+
});
178+
179+
expect(connection?.lastHealth).toMatchObject({
180+
status: "degraded",
181+
detail: expect.stringContaining("Failed connecting to MCP server"),
182+
});
183+
}),
184+
);
166185
});
167186

168187
// ---------------------------------------------------------------------------
Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
import { describe, expect, it } from "@effect/vitest";
2+
import { Effect, Layer } from "effect";
3+
import {
4+
HttpClient,
5+
HttpClientError,
6+
HttpClientRequest,
7+
HttpClientResponse,
8+
} from "effect/unstable/http";
9+
10+
import { createMcpConnector } from "./connection";
11+
12+
const endpoint = "https://internal.example/mcp";
13+
14+
describe("MCP remote transport failures", () => {
15+
it.effect("surfaces TLS verification failures without retrying SSE", () =>
16+
Effect.gen(function* () {
17+
const requests: string[] = [];
18+
const httpClientLayer = Layer.succeed(HttpClient.HttpClient)(
19+
HttpClient.make((request: HttpClientRequest.HttpClientRequest) => {
20+
requests.push(request.url);
21+
return Effect.fail(
22+
new HttpClientError.HttpClientError({
23+
reason: new HttpClientError.TransportError({
24+
request,
25+
cause: {
26+
code: "SELF_SIGNED_CERT_IN_CHAIN",
27+
message: "do-not-leak: internal certificate detail",
28+
},
29+
}),
30+
}),
31+
);
32+
}),
33+
);
34+
35+
const failure = yield* createMcpConnector({
36+
transport: "remote",
37+
endpoint,
38+
remoteTransport: "auto",
39+
httpClientLayer,
40+
}).pipe(Effect.flip);
41+
42+
expect(failure).toMatchObject({
43+
_tag: "McpConnectionError",
44+
transport: "streamable-http",
45+
failureKind: "tls",
46+
message:
47+
"MCP HTTPS connection failed: TLS certificate verification failed. Check the server certificate and Executor's CA trust configuration.",
48+
});
49+
expect(failure.message).not.toContain("do-not-leak");
50+
expect(requests).toEqual([endpoint]);
51+
}),
52+
);
53+
54+
it.effect("reports both attempts when a protocol mismatch falls back to SSE", () =>
55+
Effect.gen(function* () {
56+
const requests: string[] = [];
57+
const httpClientLayer = Layer.succeed(HttpClient.HttpClient)(
58+
HttpClient.make((request: HttpClientRequest.HttpClientRequest) => {
59+
requests.push(`${request.method} ${request.url}`);
60+
return Effect.succeed(
61+
HttpClientResponse.fromWeb(
62+
request,
63+
new Response("unsupported MCP transport", { status: 405 }),
64+
),
65+
);
66+
}),
67+
);
68+
69+
const failure = yield* createMcpConnector({
70+
transport: "remote",
71+
endpoint,
72+
remoteTransport: "auto",
73+
httpClientLayer,
74+
}).pipe(Effect.flip);
75+
76+
expect(failure).toMatchObject({
77+
_tag: "McpConnectionError",
78+
transport: "auto",
79+
failureKind: "protocol",
80+
message: "MCP auto transport failed. Streamable HTTP: HTTP 405. SSE fallback: HTTP 405.",
81+
});
82+
expect(requests.length).toBeGreaterThanOrEqual(2);
83+
}),
84+
);
85+
});

0 commit comments

Comments
 (0)