fix(selfhost): refresh stale connection tools concurrently and expose EXECUTOR_TOOLS_SYNC_TTL_MS

Author: Adityakk9031

apps/host-selfhost/src/config.ts

@@ -43,6 +43,8 @@ export interface SelfHostConfig {
   readonly organizationName: string;
   /** URL slug for org-prefixed console paths (`/<slug>/policies`). */
   readonly orgSlug: string;
+  /** Freshness TTL (in ms) for remote tool catalogs, or `null` to disable. */
+  readonly toolsSyncTtlMs?: number | null;
 }
 
 export const resolveDataDir = (): string =>
@@ -148,6 +150,7 @@ export const loadConfig = (): SelfHostConfig => {
     bootstrapAdminName: process.env.EXECUTOR_BOOTSTRAP_ADMIN_NAME ?? "Admin",
     organizationName: process.env.EXECUTOR_ORG_NAME ?? "Default",
     orgSlug: resolveOrgSlug(),
+    toolsSyncTtlMs: resolveToolsSyncTtlMs(),
   };
 };
 
@@ -165,3 +168,11 @@ const resolveOrgSlug = (): string => {
   }
   return slug;
 };
+
+const resolveToolsSyncTtlMs = (): number | null | undefined => {
+  const raw = process.env.EXECUTOR_TOOLS_SYNC_TTL_MS?.trim();
+  if (!raw) return undefined;
+  if (raw === "null" || raw === "false" || raw === "0") return null;
+  const parsed = Number.parseInt(raw, 10);
+  return Number.isNaN(parsed) ? undefined : parsed;
+};

apps/host-selfhost/src/execution.ts

@@ -55,6 +55,7 @@ export const SelfHostHostConfig: Layer.Layer<HostConfig> = Layer.sync(HostConfig
     allowLocalNetwork: config.allowLocalNetwork,
     webBaseUrl: config.webBaseUrl,
     oauthCallbackPath: "/api/oauth/callback",
+    toolsSyncTtlMs: config.toolsSyncTtlMs,
     onIntegrationChange: (event) =>
       selfHostAnalytics.record(
         event.kind === "added" ? "integration_added" : "integration_removed",

apps/host-selfhost/src/executor-config.test.ts

@@ -1,11 +1,14 @@
 import { afterEach, beforeEach, expect, test } from "@effect/vitest";
 
+import { loadConfig } from "./config";
 import executorConfig from "../executor.config";
 
 const ENV_NAME = "EXECUTOR_ALLOW_STDIO_MCP";
 const SECRET_ENV_NAME = "EXECUTOR_SECRET_KEY";
+const TTL_ENV_NAME = "EXECUTOR_TOOLS_SYNC_TTL_MS";
 const originalValue = process.env[ENV_NAME];
 const originalSecret = process.env[SECRET_ENV_NAME];
+const originalTtl = process.env[TTL_ENV_NAME];
 
 beforeEach(() => {
   process.env[SECRET_ENV_NAME] = originalSecret ?? "executor-config-test-secret";
@@ -22,6 +25,11 @@ afterEach(() => {
   } else {
     process.env[SECRET_ENV_NAME] = originalSecret;
   }
+  if (originalTtl === undefined) {
+    delete process.env[TTL_ENV_NAME];
+  } else {
+    process.env[TTL_ENV_NAME] = originalTtl;
+  }
 });
 
 const allowStdio = (): boolean => {
@@ -57,3 +65,20 @@ test("stdio MCP is enabled when the opt-in is exactly true", () => {
   process.env[ENV_NAME] = "true";
   expect(allowStdio()).toBe(true);
 });
+
+test("toolsSyncTtlMs parses integer, null/false/0 disable values, and undefined fallback", () => {
+  delete process.env[TTL_ENV_NAME];
+  expect(loadConfig().toolsSyncTtlMs).toBeUndefined();
+
+  process.env[TTL_ENV_NAME] = "60000";
+  expect(loadConfig().toolsSyncTtlMs).toBe(60000);
+
+  process.env[TTL_ENV_NAME] = "null";
+  expect(loadConfig().toolsSyncTtlMs).toBeNull();
+
+  process.env[TTL_ENV_NAME] = "false";
+  expect(loadConfig().toolsSyncTtlMs).toBeNull();
+
+  process.env[TTL_ENV_NAME] = "0";
+  expect(loadConfig().toolsSyncTtlMs).toBeNull();
+});

packages/core/api/src/server/scoped-executor.ts

@@ -97,6 +97,11 @@ export interface HostConfigShape {
    * Hosts that record product analytics supply it; omitted -> no observation.
    */
   readonly onIntegrationChange?: ExecutorConfig["onIntegrationChange"];
+  /**
+   * Freshness TTL (in ms) for remote tool catalogs before an explicit re-sync is
+   * attempted. Omit for default (15 mins), or set `null` to disable time-based re-sync.
+   */
+  readonly toolsSyncTtlMs?: number | null;
 }
 
 export class HostConfig extends Context.Service<HostConfig, HostConfigShape>()(
@@ -284,6 +289,7 @@ export const makeScopedExecutor = <
       httpClientLayer,
       fetch: hostedFetch,
       onIntegrationChange: config.onIntegrationChange,
+      ...(config.toolsSyncTtlMs !== undefined ? { toolsSyncTtlMs: config.toolsSyncTtlMs } : {}),
       onElicitation: "accept-all",
       redirectUri,
       oauthCallbackStateOrgSlug: orgSlug,

packages/core/sdk/src/executor.ts

@@ -3631,6 +3631,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
             ? b.isNull("tools_synced_at")
             : b.or(b.isNull("tools_synced_at"), b("tools_synced_at", "<", staleBefore)),
       });
+      const tasks = [];
       for (const connection of connections) {
         const integrationRow = integrationBySlug.get(connection.integration);
         if (!integrationRow) continue;
@@ -3657,24 +3658,29 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
           syncedAt < cutoff;
         if (!staleMarked && !configRevised && !expired) continue;
 
-        yield* produceConnectionTools(
-          integrationRow,
-          {
-            owner: connection.owner as Owner,
-            integration: IntegrationSlug.make(connection.integration),
-            name: ConnectionName.make(connection.name),
-          },
-          "background",
-        ).pipe(
-          Effect.catch(() => Effect.succeed([] as readonly Tool[])),
-          Effect.withSpan("executor.tools.sync_stale", {
-            attributes: {
-              "executor.integration": connection.integration,
-              "executor.connection": connection.name,
+        tasks.push(
+          produceConnectionTools(
+            integrationRow,
+            {
+              owner: connection.owner as Owner,
+              integration: IntegrationSlug.make(connection.integration),
+              name: ConnectionName.make(connection.name),
             },
-          }),
+            "background",
+          ).pipe(
+            Effect.catch(() => Effect.succeed([] as readonly Tool[])),
+            Effect.withSpan("executor.tools.sync_stale", {
+              attributes: {
+                "executor.integration": connection.integration,
+                "executor.connection": connection.name,
+              },
+            }),
+          ),
         );
       }
+      if (tasks.length > 0) {
+        yield* Effect.all(tasks, { concurrency: 10 });
+      }
     });
 
     const toolsList = (filter?: ToolListFilter): Effect.Effect<readonly Tool[], StorageFailure> =>