Single-flight managed plugin worker recovery

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts

@@ -0,0 +1,216 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { PaperclipPluginManifestV1 } from "@paperclipai/shared";
+import { createManagedBundledPluginWorkerRecovery } from "../app.js";
+import { listReadyPluginEnvironmentDrivers } from "../services/plugin-environment-driver.js";
+import type { PluginLoader } from "../services/plugin-loader.js";
+import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
+
+const mockRegistry = vi.hoisted(() => ({
+  list: vi.fn(),
+}));
+
+vi.mock("../services/plugin-registry.js", () => ({
+  pluginRegistryService: () => mockRegistry,
+}));
+
+const PLUGIN_ID = "plugin-daytona";
+const PLUGIN_KEY = "paperclip.daytona-sandbox-provider";
+
+const manifest: PaperclipPluginManifestV1 = {
+  id: PLUGIN_KEY,
+  apiVersion: 1,
+  version: "1.0.0",
+  displayName: "Daytona Sandbox Provider",
+  description: "Provides Daytona-backed sandboxes.",
+  author: "Paperclip",
+  categories: ["automation"],
+  capabilities: ["environment.drivers.register"],
+  entrypoints: { worker: "dist/worker.js" },
+  environmentDrivers: [
+    {
+      driverKey: "daytona",
+      kind: "sandbox_provider",
+      displayName: "Daytona",
+      description: "Daytona sandbox provider",
+      configSchema: { type: "object", properties: {} },
+    },
+  ],
+};
+
+function createPlugin(status: string) {
+  return {
+    id: PLUGIN_ID,
+    pluginKey: PLUGIN_KEY,
+    status,
+    manifestJson: manifest,
+  };
+}
+
+function createWorkerManager(options: {
+  running?: boolean;
+  hasHandle?: boolean;
+} = {}) {
+  let running = options.running ?? false;
+  const workerManager = {
+    isRunning: vi.fn(() => running),
+    getWorker: vi.fn(() => (options.hasHandle ? { status: "backoff" } : undefined)),
+  } as unknown as PluginWorkerManager;
+  return {
+    workerManager,
+    markRunning: () => {
+      running = true;
+    },
+  };
+}
+
+function createDeferred<T = void>() {
+  let resolve!: (value?: T | PromiseLike<T>) => void;
+  let reject!: (reason?: unknown) => void;
+  const promise = new Promise<T>((resolvePromise, rejectPromise) => {
+    resolve = resolvePromise;
+    reject = rejectPromise;
+  });
+  return { promise, resolve, reject };
+}
+
+describe("listReadyPluginEnvironmentDrivers worker recovery", () => {
+  beforeEach(() => {
+    mockRegistry.list.mockReset();
+  });
+
+  it("recovers a managed bundled provider that was installed by a sibling process after this process skipped it", async () => {
+    const plugin = createPlugin("installed");
+    mockRegistry.list.mockImplementation(async () => [plugin]);
+    const worker = createWorkerManager();
+    const startWorker = vi.fn(async () => {
+      worker.markRunning();
+      return true;
+    });
+
+    await expect(
+      listReadyPluginEnvironmentDrivers({
+        db: {} as never,
+        workerManager: worker.workerManager,
+        recoverMissingWorker: {
+          pluginKeys: [PLUGIN_KEY],
+          startWorker,
+        },
+      }),
+    ).resolves.toEqual([]);
+    expect(startWorker).not.toHaveBeenCalled();
+
+    plugin.status = "ready";
+    const drivers = await listReadyPluginEnvironmentDrivers({
+      db: {} as never,
+      workerManager: worker.workerManager,
+      recoverMissingWorker: {
+        pluginKeys: [PLUGIN_KEY],
+        startWorker,
+      },
+    });
+
+    expect(startWorker).toHaveBeenCalledWith({
+      id: PLUGIN_ID,
+      pluginKey: PLUGIN_KEY,
+    });
+    expect(drivers).toEqual([
+      expect.objectContaining({
+        pluginId: PLUGIN_ID,
+        pluginKey: PLUGIN_KEY,
+        driverKey: "daytona",
+        displayName: "Daytona",
+      }),
+    ]);
+  });
+
+  it("does not lazy-start ready plugins outside the managed bundled allowlist", async () => {
+    const plugin = createPlugin("ready");
+    mockRegistry.list.mockResolvedValue([plugin]);
+    const worker = createWorkerManager();
+    const startWorker = vi.fn(async () => {
+      worker.markRunning();
+      return true;
+    });
+
+    const drivers = await listReadyPluginEnvironmentDrivers({
+      db: {} as never,
+      workerManager: worker.workerManager,
+      recoverMissingWorker: {
+        pluginKeys: ["paperclip.kubernetes-sandbox-provider"],
+        startWorker,
+      },
+    });
+
+    expect(startWorker).not.toHaveBeenCalled();
+    expect(drivers).toEqual([]);
+  });
+
+  it("leaves existing worker-manager recovery handles alone", async () => {
+    const plugin = createPlugin("ready");
+    mockRegistry.list.mockResolvedValue([plugin]);
+    const worker = createWorkerManager({ hasHandle: true });
+    const startWorker = vi.fn(async () => true);
+
+    const drivers = await listReadyPluginEnvironmentDrivers({
+      db: {} as never,
+      workerManager: worker.workerManager,
+      recoverMissingWorker: {
+        pluginKeys: [PLUGIN_KEY],
+        startWorker,
+      },
+    });
+
+    expect(startWorker).not.toHaveBeenCalled();
+    expect(drivers).toEqual([]);
+  });
+
+  it("single-flights concurrent managed bundled recovery starts", async () => {
+    const plugin = createPlugin("ready");
+    mockRegistry.list.mockResolvedValue([plugin]);
+    const worker = createWorkerManager();
+    const loadStarted = createDeferred();
+    const releaseLoad = createDeferred();
+    const loadSingle = vi.fn(async () => {
+      loadStarted.resolve();
+      await releaseLoad.promise;
+      worker.markRunning();
+      return { success: true };
+    });
+    const startWorker = createManagedBundledPluginWorkerRecovery({
+      managedBundledPluginKeys: [PLUGIN_KEY],
+      workerManager: worker.workerManager,
+      getLoader: () => ({ loadSingle }) as Pick<PluginLoader, "loadSingle">,
+    });
+
+    const first = listReadyPluginEnvironmentDrivers({
+      db: {} as never,
+      workerManager: worker.workerManager,
+      recoverMissingWorker: {
+        pluginKeys: [PLUGIN_KEY],
+        startWorker,
+      },
+    });
+    await loadStarted.promise;
+    const second = listReadyPluginEnvironmentDrivers({
+      db: {} as never,
+      workerManager: worker.workerManager,
+      recoverMissingWorker: {
+        pluginKeys: [PLUGIN_KEY],
+        startWorker,
+      },
+    });
+
+    releaseLoad.resolve();
+    const [firstDrivers, secondDrivers] = await Promise.all([first, second]);
+
+    expect(loadSingle).toHaveBeenCalledTimes(1);
+    expect(loadSingle).toHaveBeenCalledWith(PLUGIN_ID);
+    expect(firstDrivers).toEqual([
+      expect.objectContaining({
+        pluginId: PLUGIN_ID,
+        driverKey: "daytona",
+      }),
+    ]);
+    expect(secondDrivers).toEqual(firstDrivers);
+  });
+});

server/src/app.ts

@@ -64,7 +64,7 @@ import { pluginUiStaticRoutes } from "./routes/plugin-ui-static.js";
 import { readBrandedStaticIndexHtml } from "./static-index-html.js";
 import { applyUiBranding } from "./ui-branding.js";
 import { logger } from "./middleware/logger.js";
-import { DEFAULT_LOCAL_PLUGIN_DIR, pluginLoader } from "./services/plugin-loader.js";
+import { DEFAULT_LOCAL_PLUGIN_DIR, pluginLoader, type PluginLoader } from "./services/plugin-loader.js";
 import {
   SELF_HOSTED_AUTO_INSTALL_KEYS,
   ensureBundledPlugins,
@@ -148,6 +148,55 @@ export function shouldEnablePrivateHostnameGuard(opts: {
   );
 }
 
+export function createManagedBundledPluginWorkerRecovery(input: {
+  managedBundledPluginKeys: readonly string[];
+  workerManager: Pick<PluginWorkerManager, "getWorker" | "isRunning">;
+  getLoader: () => Pick<PluginLoader, "loadSingle"> | null;
+}): (plugin: { id: string; pluginKey: string }) => Promise<boolean> {
+  const recoverablePluginKeys = new Set(input.managedBundledPluginKeys);
+  const inFlightStarts = new Map<string, Promise<boolean>>();
+
+  return async (plugin) => {
+    if (!recoverablePluginKeys.has(plugin.pluginKey)) return false;
+
+    const inFlight = inFlightStarts.get(plugin.id);
+    if (inFlight) return inFlight;
+
+    const startPromise = (async () => {
+      if (input.workerManager.getWorker(plugin.id)) {
+        return input.workerManager.isRunning(plugin.id);
+      }
+
+      const loader = input.getLoader();
+      if (!loader) return false;
+
+      try {
+        const result = await loader.loadSingle(plugin.id);
+        return result.success === true || input.workerManager.isRunning(plugin.id);
+      } catch (err) {
+        logger.warn(
+          {
+            pluginId: plugin.id,
+            pluginKey: plugin.pluginKey,
+            err: err instanceof Error ? err.message : String(err),
+          },
+          "managed bundled plugin lazy worker recovery failed",
+        );
+        throw err;
+      }
+    })();
+
+    inFlightStarts.set(plugin.id, startPromise);
+    try {
+      return await startPromise;
+    } finally {
+      if (inFlightStarts.get(plugin.id) === startPromise) {
+        inFlightStarts.delete(plugin.id);
+      }
+    }
+  };
+}
+
 export async function createApp(
   db: Db,
   opts: {
@@ -238,6 +287,34 @@ export async function createApp(
 
   const hostServicesDisposers = new Map<string, () => void>();
   const workerManager = opts.pluginWorkerManager ?? createPluginWorkerManager();
+  const managedAutoInstallKeys = opts.managedPluginAutoInstall ?? null;
+  const bundledCatalogRoot =
+    opts.bundledPluginCatalogRoot ?? resolveBundledCatalogRoot(process.env);
+  const bundledPluginInstalls = resolveBundledPluginInstalls(
+    managedAutoInstallKeys ?? SELF_HOSTED_AUTO_INSTALL_KEYS,
+    {
+      catalogRoot: bundledCatalogRoot,
+      env: process.env,
+      enforceCatalogRoot: managedAutoInstallKeys !== null,
+    },
+  );
+  const managedBundledPluginKeys =
+    managedAutoInstallKeys !== null
+      ? bundledPluginInstalls.map((install) => install.pluginKey)
+      : [];
+  let runtimePluginLoader: Pick<PluginLoader, "loadSingle"> | null = null;
+  // A sibling process can install a managed bundled plugin while this process
+  // skips the mid-install row, then finish the row after this process's
+  // loadAll() pass. The capabilities route may recover only those managed
+  // bundles by starting their ready-but-unstarted worker lazily.
+  const recoverManagedBundledPluginWorker =
+    managedAutoInstallKeys !== null
+      ? createManagedBundledPluginWorkerRecovery({
+          managedBundledPluginKeys,
+          workerManager,
+          getLoader: () => runtimePluginLoader,
+        })
+      : undefined;
 
   // Mount API routes
   const api = Router();
@@ -271,7 +348,15 @@ export async function createApp(
   api.use(fileResourceRoutes(db));
   api.use(routineRoutes(db, { pluginWorkerManager: workerManager }));
   api.use(pipelineRoutes(db));
-  api.use(environmentRoutes(db, { pluginWorkerManager: workerManager }));
+  api.use(environmentRoutes(db, {
+    pluginWorkerManager: workerManager,
+    recoverMissingPluginWorker: recoverManagedBundledPluginWorker
+      ? {
+        pluginKeys: managedBundledPluginKeys,
+        startWorker: recoverManagedBundledPluginWorker,
+      }
+      : undefined,
+  }));
   api.use(executionWorkspaceRoutes(db, { pluginWorkerManager: workerManager }));
   api.use(goalRoutes(db));
   api.use(boardChatRoutes(db, { deploymentMode: opts.deploymentMode }));
@@ -380,6 +465,7 @@ export async function createApp(
       },
     },
   );
+  runtimePluginLoader = loader;
   api.use(
     toolGatewayRoutes(db, toolGateway),
   );
@@ -570,21 +656,10 @@ export async function createApp(
   // drive the key list from the control plane; self-hosted instances keep
   // the pre-existing behavior of ensuring only the kubernetes bundle.
   //
-  // Resolution below is deliberately synchronous and NOT fail-safe: an
+  // Resolution is deliberately synchronous and NOT fail-safe: an
   // unknown key or a path escaping the bundled catalog root throws out of
   // createApp so a managed instance refuses to start (positive allowlist,
   // fail closed).
-  const managedAutoInstallKeys = opts.managedPluginAutoInstall ?? null;
-  const bundledCatalogRoot =
-    opts.bundledPluginCatalogRoot ?? resolveBundledCatalogRoot(process.env);
-  const bundledPluginInstalls = resolveBundledPluginInstalls(
-    managedAutoInstallKeys ?? SELF_HOSTED_AUTO_INSTALL_KEYS,
-    {
-      catalogRoot: bundledCatalogRoot,
-      env: process.env,
-      enforceCatalogRoot: managedAutoInstallKeys !== null,
-    },
-  );
   // SAFETY: installation is fully fail-safe. Any failure
   // (missing bundle, install error, load error) is caught, logged, and
   // swallowed per plugin so the server ALWAYS finishes booting. A degraded

server/src/routes/environments.ts

@@ -47,7 +47,10 @@ import {
 } from "../services/environment-config.js";
 import { probeEnvironment } from "../services/environment-probe.js";
 import { secretService } from "../services/secrets.js";
-import { listReadyPluginEnvironmentDrivers } from "../services/plugin-environment-driver.js";
+import {
+  listReadyPluginEnvironmentDrivers,
+  type ReadyPluginWorkerRecovery,
+} from "../services/plugin-environment-driver.js";
 import { getConfiguredSecretProvider } from "../secrets/configured-provider.js";
 import { assertBoardOrgAccess, getActorInfo } from "./authz.js";
 import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
@@ -261,7 +264,10 @@ function assertNoClientPlatformProvisionedMarkers(metadata: unknown): void {
 
 export function environmentRoutes(
   db: Db,
-  options: { pluginWorkerManager?: PluginWorkerManager } = {},
+  options: {
+    pluginWorkerManager?: PluginWorkerManager;
+    recoverMissingPluginWorker?: ReadyPluginWorkerRecovery;
+  } = {},
 ) {
   const router = Router();
   const svc = environmentService(db);
@@ -602,6 +608,7 @@ export function environmentRoutes(
     const pluginDrivers = await listReadyPluginEnvironmentDrivers({
       db,
       workerManager: options.pluginWorkerManager,
+      recoverMissingWorker: options.recoverMissingPluginWorker,
     });
     res.json(getEnvironmentCapabilities(
       AGENT_ADAPTER_TYPES,

server/src/services/plugin-environment-driver.ts

@@ -30,6 +30,27 @@ import {
 import { pluginRegistryService } from "./plugin-registry.js";
 import type { PluginWorkerManager } from "./plugin-worker-manager.js";
 
+export interface ReadyPluginWorkerRecovery {
+  pluginKeys: readonly string[];
+  startWorker(plugin: { id: string; pluginKey: string }): Promise<boolean>;
+}
+
+export interface ReadyPluginEnvironmentDriver {
+  pluginId: string;
+  pluginKey: string;
+  driverKey: string;
+  displayName: string;
+  description?: string;
+  configSchema: PluginEnvironmentDriverDeclaration["configSchema"];
+  supportsReusableLeases?: PluginEnvironmentDriverDeclaration["supportsReusableLeases"];
+  supportsInteractiveSetup?: PluginEnvironmentDriverDeclaration["supportsInteractiveSetup"];
+  interactiveSetupConnectionTypes?: PluginEnvironmentDriverDeclaration["interactiveSetupConnectionTypes"];
+  supportsTemplateCapture?: PluginEnvironmentDriverDeclaration["supportsTemplateCapture"];
+  templateRefKind?: PluginEnvironmentDriverDeclaration["templateRefKind"];
+  templateConfigBinding?: PluginEnvironmentDriverDeclaration["templateConfigBinding"];
+  supportsTemplateDelete?: PluginEnvironmentDriverDeclaration["supportsTemplateDelete"];
+}
+
 export function pluginDriverProviderKey(config: Pick<PluginEnvironmentConfig, "pluginKey" | "driverKey">): string {
   return `${config.pluginKey}:${config.driverKey}`;
 }
@@ -94,30 +115,48 @@ export async function resolvePluginSandboxProviderDriverByKey(input: {
 export async function listReadyPluginEnvironmentDrivers(input: {
   db: Db;
   workerManager?: PluginWorkerManager;
+  recoverMissingWorker?: ReadyPluginWorkerRecovery;
 }) {
   if (!input.workerManager) return [];
   const pluginRegistry = pluginRegistryService(input.db);
   const plugins = await pluginRegistry.list();
-  return plugins.flatMap((plugin) => {
-    if (plugin.status !== "ready" || !input.workerManager?.isRunning(plugin.id)) return [];
-    return (plugin.manifestJson.environmentDrivers ?? [])
-      .filter((driver) => driver.kind === "sandbox_provider")
-      .map((driver) => ({
-        pluginId: plugin.id,
-        pluginKey: plugin.pluginKey,
-        driverKey: driver.driverKey,
-        displayName: driver.displayName,
-        description: driver.description,
-        configSchema: driver.configSchema,
-        supportsReusableLeases: driver.supportsReusableLeases,
-        supportsInteractiveSetup: driver.supportsInteractiveSetup,
-        interactiveSetupConnectionTypes: driver.interactiveSetupConnectionTypes,
-        supportsTemplateCapture: driver.supportsTemplateCapture,
-        templateRefKind: driver.templateRefKind,
-        templateConfigBinding: driver.templateConfigBinding,
-        supportsTemplateDelete: driver.supportsTemplateDelete,
-      }));
-  });
+  const recoverablePluginKeys = new Set(input.recoverMissingWorker?.pluginKeys ?? []);
+  const rows: ReadyPluginEnvironmentDriver[] = [];
+  for (const plugin of plugins) {
+    if (plugin.status !== "ready") continue;
+    if (!input.workerManager.isRunning(plugin.id)) {
+      const canRecover =
+        recoverablePluginKeys.has(plugin.pluginKey)
+        && !input.workerManager.getWorker(plugin.id);
+      if (canRecover) {
+        await input.recoverMissingWorker?.startWorker({
+          id: plugin.id,
+          pluginKey: plugin.pluginKey,
+        }).catch(() => false);
+      }
+      if (!input.workerManager.isRunning(plugin.id)) continue;
+    }
+    rows.push(
+      ...(plugin.manifestJson.environmentDrivers ?? [])
+        .filter((driver) => driver.kind === "sandbox_provider")
+        .map((driver) => ({
+          pluginId: plugin.id,
+          pluginKey: plugin.pluginKey,
+          driverKey: driver.driverKey,
+          displayName: driver.displayName,
+          description: driver.description,
+          configSchema: driver.configSchema,
+          supportsReusableLeases: driver.supportsReusableLeases,
+          supportsInteractiveSetup: driver.supportsInteractiveSetup,
+          interactiveSetupConnectionTypes: driver.interactiveSetupConnectionTypes,
+          supportsTemplateCapture: driver.supportsTemplateCapture,
+          templateRefKind: driver.templateRefKind,
+          templateConfigBinding: driver.templateConfigBinding,
+          supportsTemplateDelete: driver.supportsTemplateDelete,
+        })),
+    );
+  }
+  return rows;
 }
 
 export async function validatePluginSandboxProviderConfig(input: {