Fix MCP catalog refresh races and transport diagnostics

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

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) => {

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<void>();
+        const releaseRefresh = yield* Deferred.make<void>();
+        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",
     () =>

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 = <const TPlugins extends readonly AnyPlugin[] = rea
     const syncHealthReason = (result: ResolveToolsResult): string =>
       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<readonly Tool[], IntegrationNotFoundError | StorageFailure> =>
       Effect.gen(function* () {
         const runtime = runtimes.get(integrationRow.plugin_id);
@@ -2631,7 +2631,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         }
 
         if (
-          mode === "background" &&
+          mode() === "background" &&
           runtime.plugin.remoteToolCatalog === true &&
           result.tools.length === 0
         ) {
@@ -2711,6 +2711,38 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         );
       });
 
+    type ToolProductionError = IntegrationNotFoundError | StorageFailure;
+    interface ToolProductionInFlight {
+      readonly deferred: Deferred.Deferred<readonly Tool[], ToolProductionError>;
+      mode: "explicit" | "background";
+    }
+    const toolProductionInFlight = new Map<string, ToolProductionInFlight>();
+    const produceConnectionTools = (
+      integrationRow: IntegrationRow,
+      ref: ConnectionRef,
+      requestedMode: "explicit" | "background" = "explicit",
+    ): Effect.Effect<readonly Tool[], ToolProductionError> =>
+      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<readonly Tool[], ToolProductionError>(),
+          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
     // ------------------------------------------------------------------

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"),
+      });
+    }),
+  );
 });
 
 // ---------------------------------------------------------------------------

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);
+    }),
+  );
+});