From 78f8c6c3d4e8f65d53b69b5cd37861f99918fba6 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Wed, 29 Jul 2026 12:50:16 -0700 Subject: [PATCH] Recover managed bundled plugin workers on demand (#10429) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Managed deployments can auto-provision bundled sandbox-provider plugins so cloud or remote execution environments appear in the board UI > - In a multi-service deployment, several server processes can share one database and boot concurrently > - A sibling process can create a bundled plugin row while the web process sees it before it reaches `ready` > - The web process correctly avoids clobbering the existing row, but its startup `loadAll()` can miss the plugin and never start that worker locally > - The environments capabilities route then filters out the sandbox provider because the plugin is ready in the database but not running in the web process > - This pull request adds a narrow managed-bundle recovery path that lazily starts the missing worker when the capabilities route sees a ready managed bundled plugin > - The benefit is that the sandbox provider becomes visible after the install finishes, without requiring a web-process restart ## Linked Issues or Issue Description - No public GitHub issue found for this exact deployment race. - Related broad plugin runtime context: Refs #432. Bug description: - What happened: in a managed multi-service deployment with shared database state and bundled plugin auto-install enabled, the API-serving process can skip a plugin row while it is still `installed`, run startup plugin loading before that row becomes `ready`, and then permanently omit the sandbox provider from environment capabilities. - Expected behavior: once the managed bundled plugin row reaches `ready`, the API-serving process should be able to start the plugin worker and include its sandbox provider without a restart. - Steps to reproduce: boot a web process and a sibling worker process concurrently; have the sibling create the bundled plugin row and transition it to `ready` after the web process has already skipped auto-install and run `loadAll()`. - Deployment mode: managed multi-service deployment with shared database state and `plugins.autoInstall` configured. ## What Changed - Added a managed bundled plugin worker recovery helper that single-flights lazy `loadSingle()` starts and only allows configured managed bundled plugin keys. - Passed the managed recovery hook into the environments capabilities route. - Updated `listReadyPluginEnvironmentDrivers()` to attempt bounded recovery for ready managed bundled plugins whose worker is missing in the current process, and only for plugins that actually declare a `sandbox_provider` environment driver. - Made request-time recovery use `loadSingle(id, { markErrorOnFailure: false })` so a local activation failure in one process never transitions the shared plugin row to `error` (a sibling process may be running the plugin successfully). - When error writes are suppressed and activation fails after the worker was spawned, the loader now tears down the partially-registered local runtime (scheduler registration, event subscriptions, agent tools, worker process) instead of leaving a half-activated worker lingering; the teardown steps are factored out of `unloadSingle()` into a shared helper. - A failed recovery attempt now discards the crashed/stopped handle it left registered in the worker manager (a worker that dies during initialize is killed without a scheduled restart), so later capability requests can retry recovery instead of being blocked by the handle-presence gate until a process restart. Handles in starting/running/backoff states are left to the worker manager's own lifecycle; recovery only ever starts when no handle existed, so no pre-existing worker can be affected. - Added a regression test suite covering the installed-to-ready race, allowlist behavior, the driver-kind gate, existing worker handles, concurrent single-flight recovery, bounded slow recovery attempts, suppressed shared error-state writes, partial-runtime teardown on late activation failure, and retry after a dead handle is discarded. ## Verification - `pnpm vitest run src/__tests__/plugin-environment-driver-ready-recovery.test.ts` (in `server/`) passed: 10 tests. - `pnpm --filter @paperclipai/server typecheck` passed. ## Risks - Low risk for self-hosted single-process deployments because lazy recovery is only wired when managed plugin auto-install config is present; with no managed config the capabilities route takes the exact pre-change code path. - The capabilities route can wait briefly while attempting recovery; the attempt is bounded and defaults to 2 seconds. - Failed recovery keeps the prior behavior of omitting the provider until a later successful worker start, and now also cleans up any partially-started local worker so retries begin from a clean slate. ## Model Used - Initial implementation: OpenAI GPT-5 via Codex local coding agent, with repository tool use and command execution. - Review-feedback follow-ups (driver-kind gate, partial-runtime teardown, expanded regression tests): Claude Fable 5 (claude-fable-5) via Claude Code, with repository tool use and command execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- ...-environment-driver-ready-recovery.test.ts | 515 ++++++++++++++++++ server/src/app.ts | 135 ++++- server/src/routes/environments.ts | 11 +- .../src/services/plugin-environment-driver.ts | 114 +++- server/src/services/plugin-loader.ts | 162 ++++-- 5 files changed, 850 insertions(+), 87 deletions(-) create mode 100644 server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts diff --git a/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts b/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts new file mode 100644 index 0000000000..718c399153 --- /dev/null +++ b/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts @@ -0,0 +1,515 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +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 { pluginLoader, type PluginLoader } from "../services/plugin-loader.js"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; + +const mockRegistry = vi.hoisted(() => ({ + getById: vi.fn(), + list: vi.fn(), + listConfigs: vi.fn(), + update: 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, options: { id?: string; pluginKey?: string } = {}) { + return { + id: options.id ?? PLUGIN_ID, + pluginKey: options.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() { + let resolve!: (value?: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +describe("listReadyPluginEnvironmentDrivers worker recovery", () => { + beforeEach(() => { + mockRegistry.getById.mockReset(); + mockRegistry.list.mockReset(); + mockRegistry.listConfigs.mockReset(); + mockRegistry.update.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("does not lazy-start managed bundled plugins without sandbox provider drivers", async () => { + const plugin = { + ...createPlugin("ready"), + manifestJson: { + ...manifest, + capabilities: [], + environmentDrivers: undefined, + }, + }; + 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: [PLUGIN_KEY], + 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, + }); + + 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, { + markErrorOnFailure: false, + }); + expect(firstDrivers).toEqual([ + expect.objectContaining({ + pluginId: PLUGIN_ID, + driverKey: "daytona", + }), + ]); + expect(secondDrivers).toEqual(firstDrivers); + }); + + it("keeps request-time recovery failures process-local instead of marking shared plugin state errored", async () => { + const plugin = createPlugin("ready"); + mockRegistry.list.mockResolvedValue([plugin]); + const worker = createWorkerManager(); + const loadSingle = vi.fn(async () => ({ + plugin, + success: false, + registered: { worker: false, eventSubscriptions: 0, jobs: 0, webhooks: 0, tools: 0 }, + error: "local worker spawn failed", + })); + const startWorker = createManagedBundledPluginWorkerRecovery({ + managedBundledPluginKeys: [PLUGIN_KEY], + workerManager: worker.workerManager, + getLoader: () => ({ loadSingle }) as Pick, + }); + + const drivers = await listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager: worker.workerManager, + recoverMissingWorker: { + pluginKeys: [PLUGIN_KEY], + startWorker, + }, + }); + + expect(loadSingle).toHaveBeenCalledWith(PLUGIN_ID, { + markErrorOnFailure: false, + }); + expect(drivers).toEqual([]); + }); + + it("discards the dead handle a failed recovery attempt leaves behind so later requests retry", async () => { + const plugin = createPlugin("ready"); + mockRegistry.list.mockResolvedValue([plugin]); + + // Simulates the initialize-failure path: startWorker registers a handle, + // the worker dies during initialize, and no restart is scheduled — the + // crashed handle stays registered in the worker manager. + let handle: { status: string } | undefined; + let running = false; + const stopWorker = vi.fn(async () => { + handle = undefined; + }); + const workerManager = { + isRunning: vi.fn(() => running), + getWorker: vi.fn(() => handle), + stopWorker, + } as unknown as PluginWorkerManager; + + const loadSingle = vi.fn() + .mockImplementationOnce(async () => { + handle = { status: "crashed" }; + throw new Error("Worker initialize failed"); + }) + .mockImplementationOnce(async () => { + handle = { status: "running" }; + running = true; + return { plugin, success: true } as never; + }); + + const startWorker = createManagedBundledPluginWorkerRecovery({ + managedBundledPluginKeys: [PLUGIN_KEY], + workerManager, + getLoader: () => ({ loadSingle }) as Pick, + }); + + const firstDrivers = await listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager, + recoverMissingWorker: { + pluginKeys: [PLUGIN_KEY], + startWorker, + }, + }); + + expect(firstDrivers).toEqual([]); + expect(stopWorker).toHaveBeenCalledWith(PLUGIN_ID); + expect(handle).toBeUndefined(); + + const secondDrivers = await listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager, + recoverMissingWorker: { + pluginKeys: [PLUGIN_KEY], + startWorker, + }, + }); + + expect(loadSingle).toHaveBeenCalledTimes(2); + expect(secondDrivers).toEqual([ + expect.objectContaining({ + pluginId: PLUGIN_ID, + driverKey: "daytona", + }), + ]); + }); + + it("lets callers suppress shared error-state writes on activation failure", async () => { + const plugin = { + ...createPlugin("ready"), + packageName: "@paperclipai/missing-sandbox-provider", + packagePath: null, + version: "1.0.0", + }; + mockRegistry.getById.mockResolvedValue(plugin); + const lifecycleManager = { + markError: vi.fn().mockResolvedValue(undefined), + }; + const loader = pluginLoader( + {} as never, + { + enableLocalFilesystem: false, + enableNpmDiscovery: false, + localPluginDir: "__missing_plugin_dir__", + }, + { + lifecycleManager, + workerManager: {}, + eventBus: {}, + jobScheduler: {}, + jobStore: {}, + toolDispatcher: {}, + buildHostHandlers: vi.fn(), + instanceInfo: { + instanceId: "test", + hostVersion: "0.0.0", + }, + } as never, + ); + + await expect(loader.loadSingle(PLUGIN_ID)).resolves.toMatchObject({ + success: false, + error: expect.stringContaining("Package root not found"), + }); + expect(lifecycleManager.markError).toHaveBeenCalledTimes(1); + + lifecycleManager.markError.mockClear(); + await expect(loader.loadSingle(PLUGIN_ID, { markErrorOnFailure: false })).resolves.toMatchObject({ + success: false, + error: expect.stringContaining("Package root not found"), + }); + expect(lifecycleManager.markError).not.toHaveBeenCalled(); + }); + + it("tears down partially-activated runtime state when error writes are suppressed", async () => { + const fixtureDir = fs.mkdtempSync(path.join(os.tmpdir(), "plugin-recovery-fixture-")); + try { + fs.writeFileSync( + path.join(fixtureDir, "package.json"), + JSON.stringify({ + name: "@paperclipai/daytona-sandbox-provider", + version: "1.0.0", + paperclipPlugin: { manifest: "manifest.mjs" }, + }), + ); + fs.writeFileSync( + path.join(fixtureDir, "manifest.mjs"), + `export default ${JSON.stringify(manifest)};`, + ); + fs.mkdirSync(path.join(fixtureDir, "dist")); + fs.writeFileSync(path.join(fixtureDir, "dist", "worker.js"), ""); + + const plugin = { + ...createPlugin("ready"), + packageName: "@paperclipai/daytona-sandbox-provider", + packagePath: fixtureDir, + version: "1.0.0", + }; + mockRegistry.getById.mockResolvedValue(plugin); + mockRegistry.listConfigs.mockResolvedValue([]); + mockRegistry.update.mockResolvedValue(undefined); + + const lifecycleManager = { markError: vi.fn().mockResolvedValue(undefined) }; + const workerManager = { + startWorker: vi.fn().mockResolvedValue(undefined), + isRunning: vi.fn(() => true), + stopWorker: vi.fn().mockResolvedValue(undefined), + }; + const eventBus = { + forPlugin: vi.fn(() => { + throw new Error("event bus offline"); + }), + clearPlugin: vi.fn(), + subscriptionCount: vi.fn(() => 0), + }; + const jobScheduler = { unregisterPlugin: vi.fn().mockResolvedValue(undefined) }; + const toolDispatcher = { unregisterPluginTools: vi.fn(), registerPluginTools: vi.fn() }; + + const loader = pluginLoader( + {} as never, + { + enableLocalFilesystem: false, + enableNpmDiscovery: false, + localPluginDir: "__missing_plugin_dir__", + }, + { + lifecycleManager, + workerManager, + eventBus, + jobScheduler, + jobStore: {}, + toolDispatcher, + buildHostHandlers: vi.fn(() => ({})), + instanceInfo: { + instanceId: "test", + hostVersion: "0.0.0", + }, + } as never, + ); + + await expect(loader.loadSingle(PLUGIN_ID, { markErrorOnFailure: false })).resolves.toMatchObject({ + success: false, + error: expect.stringContaining("event bus offline"), + }); + + expect(lifecycleManager.markError).not.toHaveBeenCalled(); + expect(jobScheduler.unregisterPlugin).toHaveBeenCalledWith(PLUGIN_ID); + expect(eventBus.clearPlugin).toHaveBeenCalledWith(PLUGIN_KEY); + expect(toolDispatcher.unregisterPluginTools).toHaveBeenCalledWith(PLUGIN_KEY); + expect(workerManager.stopWorker).toHaveBeenCalledWith(PLUGIN_ID); + } finally { + fs.rmSync(fixtureDir, { recursive: true, force: true }); + } + }); + + it("bounds slow managed bundled recovery attempts without serial waits", async () => { + vi.useFakeTimers(); + try { + const plugins = [ + createPlugin("ready"), + createPlugin("ready", { + id: "plugin-modal", + pluginKey: "paperclip.modal-sandbox-provider", + }), + ]; + mockRegistry.list.mockResolvedValue(plugins); + const worker = createWorkerManager(); + const startWorker = vi.fn(() => new Promise(() => {})); + + const driversPromise = listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager: worker.workerManager, + recoverMissingWorker: { + pluginKeys: [PLUGIN_KEY, "paperclip.modal-sandbox-provider"], + startWorker, + timeoutMs: 25, + }, + }); + await Promise.resolve(); + + expect(startWorker).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(25); + + await expect(driversPromise).resolves.toEqual([]); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/server/src/app.ts b/server/src/app.ts index 5eeeb751c0..5b6f5d529d 100644 --- a/server/src/app.ts +++ b/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,87 @@ export function shouldEnablePrivateHostnameGuard(opts: { ); } +export function createManagedBundledPluginWorkerRecovery(input: { + managedBundledPluginKeys: readonly string[]; + workerManager: Pick; + getLoader: () => Pick | null; +}): (plugin: { id: string; pluginKey: string }) => Promise { + const recoverablePluginKeys = new Set(input.managedBundledPluginKeys); + const inFlightStarts = new Map>(); + + // A failed attempt can leave behind the dead handle it registered (e.g. the + // worker process died during initialize, which kills the process without + // scheduling a restart). No pre-existing handle survives to a recovery + // attempt — recovery only starts when getWorker() was empty — so discarding + // the dead handle lets a later capability request retry instead of being + // blocked by the handle-presence gate until the process restarts. Handles + // in starting/running/backoff states belong to the worker manager's own + // lifecycle and are left alone. + const discardDeadRecoveryHandle = async (plugin: { id: string; pluginKey: string }) => { + const handle = input.workerManager.getWorker(plugin.id); + if (!handle || (handle.status !== "crashed" && handle.status !== "stopped")) return; + try { + await input.workerManager.stopWorker(plugin.id); + } catch (err) { + logger.warn( + { + pluginId: plugin.id, + pluginKey: plugin.pluginKey, + err: err instanceof Error ? err.message : String(err), + }, + "failed to discard dead plugin worker handle after recovery failure", + ); + } + }; + + 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, { + markErrorOnFailure: false, + }); + if (result.success === true || input.workerManager.isRunning(plugin.id)) { + return true; + } + await discardDeadRecoveryHandle(plugin); + return false; + } 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", + ); + await discardDeadRecoveryHandle(plugin); + 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 +319,34 @@ export async function createApp( const hostServicesDisposers = new Map 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 | 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 +380,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 +497,7 @@ export async function createApp( }, }, ); + runtimePluginLoader = loader; api.use( toolGatewayRoutes(db, toolGateway), ); @@ -570,21 +688,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 diff --git a/server/src/routes/environments.ts b/server/src/routes/environments.ts index e26f687ccc..e83cc07c18 100644 --- a/server/src/routes/environments.ts +++ b/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, diff --git a/server/src/services/plugin-environment-driver.ts b/server/src/services/plugin-environment-driver.ts index 72d2425f19..077864bc84 100644 --- a/server/src/services/plugin-environment-driver.ts +++ b/server/src/services/plugin-environment-driver.ts @@ -30,10 +30,49 @@ 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; + timeoutMs?: number; +} + +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): string { return `${config.pluginKey}:${config.driverKey}`; } +const DEFAULT_READY_PLUGIN_WORKER_RECOVERY_TIMEOUT_MS = 2_000; + +async function resolveWithTimeout(promise: Promise, timeoutMs: number, timeoutValue: T): Promise { + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return await promise; + let timeout: ReturnType | null = null; + try { + return await Promise.race([ + promise, + new Promise((resolve) => { + timeout = setTimeout(() => resolve(timeoutValue), timeoutMs); + }), + ]); + } finally { + if (timeout) clearTimeout(timeout); + } +} + export async function resolvePluginEnvironmentDriver(input: { db: Db; workerManager: PluginWorkerManager; @@ -94,30 +133,67 @@ 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, + const recoverablePluginKeys = new Set(input.recoverMissingWorker?.pluginKeys ?? []); + const readyPlugins = plugins.filter((plugin) => plugin.status === "ready"); + const recoveryAttempts: Promise[] = []; + + for (const plugin of readyPlugins) { + const hasSandboxProviderDriver = plugin.manifestJson.environmentDrivers?.some( + (driver) => driver.kind === "sandbox_provider", + ) ?? false; + const canRecover = + hasSandboxProviderDriver + && !input.workerManager.isRunning(plugin.id) + && recoverablePluginKeys.has(plugin.pluginKey) + && !input.workerManager.getWorker(plugin.id); + if (!canRecover || !input.recoverMissingWorker) continue; + const timeoutMs = + input.recoverMissingWorker.timeoutMs ?? DEFAULT_READY_PLUGIN_WORKER_RECOVERY_TIMEOUT_MS; + recoveryAttempts.push(resolveWithTimeout( + input.recoverMissingWorker.startWorker({ + id: 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, - })); - }); + }).catch(() => false), + timeoutMs, + false, + )); + } + + if (recoveryAttempts.length > 0) { + await Promise.all(recoveryAttempts); + } + + const rows: ReadyPluginEnvironmentDriver[] = []; + for (const plugin of readyPlugins) { + 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: { diff --git a/server/src/services/plugin-loader.ts b/server/src/services/plugin-loader.ts index 5df3f62cf6..b2079a93e4 100644 --- a/server/src/services/plugin-loader.ts +++ b/server/src/services/plugin-loader.ts @@ -360,6 +360,20 @@ export interface PluginLoadResult { }; } +export interface PluginLoadSingleOptions { + /** + * When false, activation failures are reported to the caller but do not + * transition the shared plugin row to `error`. Partially-registered local + * runtime state (worker process, scheduler/tool registrations) is still + * torn down in this process. + */ + markErrorOnFailure?: boolean; +} + +interface PluginActivateOptions { + markErrorOnFailure: boolean; +} + /** * Result of activating all ready plugins at server startup. */ @@ -540,7 +554,7 @@ export interface PluginLoader { * * @see PLUGIN_SPEC.md §8.3 — Install Process */ - loadSingle(pluginId: string): Promise; + loadSingle(pluginId: string, options?: PluginLoadSingleOptions): Promise; /** * Deactivate a single plugin — stop its worker and unregister all @@ -1937,9 +1951,10 @@ export function pluginLoader( * capabilities (tools, jobs, etc.). * * @param pluginId - The UUID of the plugin to load. + * @param options - Optional activation behavior overrides. * @returns A promise that resolves with the result of the activation. */ - async loadSingle(pluginId: string): Promise { + async loadSingle(pluginId: string, options: PluginLoadSingleOptions = {}): Promise { if (!runtimeServices) { throw new Error( "Cannot loadSingle: no PluginRuntimeServices provided. " + @@ -1975,7 +1990,9 @@ export function pluginLoader( ); } - return activatePlugin(plugin); + return activatePlugin(plugin, { + markErrorOnFailure: options.markErrorOnFailure ?? true, + }); }, // ----------------------------------------------------------------------- @@ -1994,40 +2011,7 @@ export function pluginLoader( "plugin-loader: unloading single plugin", ); - const { - workerManager, - eventBus, - jobScheduler, - toolDispatcher, - } = runtimeServices; - - // 1. Unregister from job scheduler (cancels in-flight runs) - try { - await jobScheduler.unregisterPlugin(pluginId); - } catch (err) { - log.warn( - { pluginId, err: err instanceof Error ? err.message : String(err) }, - "plugin-loader: failed to unregister from job scheduler (best-effort)", - ); - } - - // 2. Clear event subscriptions - eventBus.clearPlugin(pluginKey); - - // 3. Unregister agent tools - toolDispatcher.unregisterPluginTools(pluginKey); - - // 4. Stop the worker process - try { - if (workerManager.isRunning(pluginId)) { - await workerManager.stopWorker(pluginId); - } - } catch (err) { - log.warn( - { pluginId, err: err instanceof Error ? err.message : String(err) }, - "plugin-loader: failed to stop worker during unload (best-effort)", - ); - } + await teardownPluginRuntime(pluginId, pluginKey); log.info( { pluginId, pluginKey }, @@ -2060,6 +2044,58 @@ export function pluginLoader( }, }; + // ------------------------------------------------------------------------- + // Internal: teardownPluginRuntime — shared by unloadSingle and activation + // failure cleanup + // ------------------------------------------------------------------------- + + /** + * Tear down a plugin's runtime state in this process: scheduler + * registration, event subscriptions, agent tools, and the worker process. + * Does not touch the plugin's database row. + */ + async function teardownPluginRuntime( + pluginId: string, + pluginKey: string, + ): Promise { + if (!runtimeServices) return; + + const { + workerManager, + eventBus, + jobScheduler, + toolDispatcher, + } = runtimeServices; + + // 1. Unregister from job scheduler (cancels in-flight runs) + try { + await jobScheduler.unregisterPlugin(pluginId); + } catch (err) { + log.warn( + { pluginId, err: err instanceof Error ? err.message : String(err) }, + "plugin-loader: failed to unregister from job scheduler (best-effort)", + ); + } + + // 2. Clear event subscriptions + eventBus.clearPlugin(pluginKey); + + // 3. Unregister agent tools + toolDispatcher.unregisterPluginTools(pluginKey); + + // 4. Stop the worker process + try { + if (workerManager.isRunning(pluginId)) { + await workerManager.stopWorker(pluginId); + } + } catch (err) { + log.warn( + { pluginId, err: err instanceof Error ? err.message : String(err) }, + "plugin-loader: failed to stop worker during unload (best-effort)", + ); + } + } + // ------------------------------------------------------------------------- // Internal: activatePlugin — shared logic for loadAll and loadSingle // ------------------------------------------------------------------------- @@ -2069,10 +2105,13 @@ export function pluginLoader( * sync jobs, register tools. * * This is the core orchestration logic shared by `loadAll()` and `loadSingle()`. - * Failures are caught and reported in the result; the plugin is marked as - * `error` in the database when activation fails. + * Failures are caught and reported in the result. By default the plugin is + * marked as `error` in the database when activation fails. */ - async function activatePlugin(plugin: PluginRecord): Promise { + async function activatePlugin( + plugin: PluginRecord, + options: PluginActivateOptions = { markErrorOnFailure: true }, + ): Promise { const pluginId = plugin.id; const pluginKey = plugin.pluginKey; let activePlugin = plugin; @@ -2351,18 +2390,37 @@ export function pluginLoader( "plugin-loader: failed to activate plugin", ); - // Mark the plugin as errored in the database so it is not retried - // automatically on next startup without operator intervention. - try { - await lifecycleManager.markError(pluginId, `Activation failed: ${errorMessage}`); - } catch (markErr) { - log.error( - { - pluginId, - err: markErr instanceof Error ? markErr.message : String(markErr), - }, - "plugin-loader: failed to mark plugin as error after activation failure", - ); + if (options.markErrorOnFailure) { + // Mark the plugin as errored in the database so it is not retried + // automatically on next startup without operator intervention. + // markError also deactivates the plugin runtime in this process. + try { + await lifecycleManager.markError(pluginId, `Activation failed: ${errorMessage}`); + } catch (markErr) { + log.error( + { + pluginId, + err: markErr instanceof Error ? markErr.message : String(markErr), + }, + "plugin-loader: failed to mark plugin as error after activation failure", + ); + } + } else if (registered.worker) { + // The shared plugin row stays untouched, but this process spawned a + // worker before the failure — tear down the partially-registered + // runtime so a half-activated plugin does not linger locally. + try { + await teardownPluginRuntime(pluginId, pluginKey); + } catch (cleanupErr) { + log.warn( + { + pluginId, + pluginKey, + err: cleanupErr instanceof Error ? cleanupErr.message : String(cleanupErr), + }, + "plugin-loader: failed to tear down partially-activated plugin runtime", + ); + } } return {